specs/ARCHITECTURE.md

# ARCHITECTURE.md — Fugee

> System architecture, data flow, and component contracts.
> Read `CLAUDE.md` first, then this file, then `PLAN.md`.

---

## System Overview

Fugee is a **single-process Python application**: a Gradio frontend and a
pure-Python agent loop running in the same process. There is no subprocess,
no Node.js, no NDJSON bridge.

The agent loop is ported from the structural patterns of pi-agent-core
(the loop, event stream, tool lifecycle, steering queue) but implemented
entirely in Python. LLM calls go directly to Ollama or any OpenAI-compatible
endpoint via the `ollama` or `litellm` Python SDK.

```
┌─────────────────────────────────────────────────────────┐
│                     HF Space / Browser                   │
│  ┌───────────────────────────────────────────────────┐   │
│  │                  Gradio App (Python)               │   │
│  │                                                    │   │
│  │  Phase 1   Phase 2   Phase 3   Phase 4   Phase 5  │   │
│  │  Intake    Interview  Assessment  Recs    Docs     │   │
│  │     │          │          │         │       │      │   │
│  │     └──────────┴──────────┴─────────┴───────┘      │   │
│  │                      │                              │   │
│  │              AgentLoop (agent/loop.py)              │   │
│  │         async generator — yields AgentEvent         │   │
│  └──────────────────────┼──────────────────────────────┘   │
│                         │                                   │
│  ┌──────────────────────▼──────────────────────────────┐   │
│  │              Python Agent Loop                       │   │
│  │              agent/loop.py                          │   │
│  │                                                      │   │
│  │  ┌─────────────┐   ┌──────────────┐                 │   │
│  │  │  LLM client │   │  AgentTools  │                 │   │
│  │  │  (ollama /  │   │  web_search  │                 │   │
│  │  │   litellm)  │   │  country_lookup│               │   │
│  │  └──────┬──────┘   └──────┬───────┘                 │   │
│  └─────────┼─────────────────┼─────────────────────────┘   │
│            │                 │                               │
│    ┌───────▼──────┐  ┌───────▼──────────┐                   │
│    │  LLM endpoint│  │  External APIs   │                   │
│    │  (Ollama /   │  │  (UNHCR, asylum  │                   │
│    │   Modal)     │  │   policy search) │                   │
│    └──────────────┘  └──────────────────┘                   │
└─────────────────────────────────────────────────────────────┘
```

---

## Design Decision: Why Pure Python

The original spec called for a pi-agent-core (Node.js) subprocess bridge.
After evaluation, we ported the agent loop to Python instead. Reasons:

1. **HF Spaces constraint** — spawning a Node subprocess on a Gradio Space
   is fragile (subprocess lifecycle, signal handling, buffering in sandbox).
2. **Single process** — Gradio's `async` generator support means the loop
   can yield events directly into the UI with no IPC layer.
3. **Porting cost is low** — the pi-agent-core loop is ~300 lines of logic.
   Python's `asyncio`, `typing`, and `dataclasses` map cleanly.
4. **LLM layer is commodity** — `ollama` Python SDK covers the Ollama case
   in 2 lines; `litellm` covers multi-provider if needed. No need to port
   pi-ai's 20-provider abstraction.

What was kept from pi-agent-core's design:
- The while-loop with tool execution and event emission pattern
- Typed event contracts (ported to Python dataclasses)
- Steering queue (inject messages mid-run)
- Tool definition schema (ported to Python TypedDict / Pydantic)

**Source:** https://github.com/earendil-works/pi/tree/main/packages/pi-agent-core/src
Read `agent-loop.ts`, `event-stream.ts`, and `types.ts` before implementing `agent/loop.py`.

---

## Component Contracts

### 1. Agent Loop (`agent/loop.py`)

Pure Python async generator. Gradio phases call `loop.run(prompt, session)`
and iterate over the yielded `AgentEvent` objects.

**Event types** (Python dataclasses in `agent/events.py`):

```python
@dataclass
class AgentStartEvent:    type: str = "agent_start"
@dataclass
class TurnStartEvent:     type: str = "turn_start"
@dataclass
class TextDeltaEvent:     type: str = "text_delta";  delta: str = ""
@dataclass
class ToolStartEvent:     type: str = "tool_start";  name: str = ""; args: dict = field(default_factory=dict)
@dataclass
class ToolEndEvent:       type: str = "tool_end";    name: str = ""; result: dict = field(default_factory=dict)
@dataclass
class TurnEndEvent:       type: str = "turn_end";    message: dict = field(default_factory=dict)
@dataclass
class AgentEndEvent:      type: str = "agent_end";   messages: list = field(default_factory=list)
@dataclass
class ErrorEvent:         type: str = "error";       message: str = ""

AgentEvent = Union[AgentStartEvent, TurnStartEvent, TextDeltaEvent,
                   ToolStartEvent, ToolEndEvent, TurnEndEvent,
                   AgentEndEvent, ErrorEvent]
```

**Loop contract:**

```python
async def run(
    prompt: str,
    session: SessionState,
    system_prompt: str,
    tools: list[AgentTool],
    thinking_level: Literal["low", "medium", "high"] = "low",
) -> AsyncGenerator[AgentEvent, None]:
    ...
```

- MUST NOT buffer; yield each event as it is produced
- MUST propagate errors as `ErrorEvent` (never raise through Gradio)
- MUST support `steering_queue` injection mid-run (for follow-up questions)
- MUST support `abort_event` to stop the loop gracefully

**Loop internals (ported from pi-agent-core patterns):**

```python
# Simplified structure — see agent/loop.py for full implementation
async def run(...):
    yield AgentStartEvent()
    messages = list(session.messages)
    messages.append({"role": "user", "content": prompt})

    while True:
        yield TurnStartEvent()
        response = await llm.stream(messages, tools, system_prompt)

        # Stream text deltas
        async for chunk in response:
            if chunk.type == "text":
                yield TextDeltaEvent(delta=chunk.text)
            elif chunk.type == "tool_call":
                yield ToolStartEvent(name=chunk.name, args=chunk.args)
                result = await execute_tool(chunk.name, chunk.args, tools)
                yield ToolEndEvent(name=chunk.name, result=result)

        # Check steering queue
        if not steering_queue.empty():
            steer_msg = await steering_queue.get()
            messages.append({"role": "user", "content": steer_msg})
            continue

        yield TurnEndEvent(message=response.final_message)
        break

    yield AgentEndEvent(messages=messages)
```

---

### 2. Agent Tools (`agent/tools/`)

Tools are Python callables registered with the loop. Each tool is a
`AgentTool` dataclass:

```python
@dataclass
class AgentTool:
    name: str
    description: str
    parameters: dict           # JSON Schema object
    execute: Callable          # async (args: dict) -> dict
```

**`web_search` tool** (`agent/tools/web_search.py`):
```python
AgentTool(
    name="web_search",
    description="Search for current asylum policies, UNHCR data, country safety",
    parameters={
        "type": "object",
        "properties": {
            "query": {"type": "string"},
            "focus": {"type": "string", "enum": ["asylum", "safety", "process", "contacts"]}
        },
        "required": ["query"]
    },
    execute=web_search_execute   # calls Tavily or Serper API
)
```

**`country_lookup` tool** (`agent/tools/country_lookup.py`):
```python
AgentTool(
    name="country_lookup",
    description="Look up a country's asylum program: acceptance rates, processing times, UNHCR presence",
    parameters={
        "type": "object",
        "properties": {
            "country": {"type": "string"},
            "profile": {
                "type": "object",
                "properties": {
                    "origin": {"type": "string"},
                    "persecutionType": {"type": "string"}
                }
            }
        },
        "required": ["country"]
    },
    # Data source: specs/data/countries_enriched.json (preferred, post enrich_downloader.py)
    #              specs/data/countries.json           (curated fallback, always present)
    # Both share the same schema — see specs/data/README.md.
    # Load at startup; do NOT re-read per call. Fail loudly if neither file exists.
    execute=country_lookup_execute
)
```

---

### 3. Session State Machine (`app/state/session.py`)

The interview follows a strict state machine. States are append-only — no
going back to modify earlier answers once the user has moved forward.

```
LANGUAGE_SELECT
      │
      ▼
INTAKE          ← Phase 1 complete
      │
      ▼
SITUATION       ─ "What type of persecution?"
      │            "Are you in immediate danger?"
      │            "Current location / transit countries?"
      ▼
HISTORY         ─ "How long have you been displaced?"
      │            "Have you made prior asylum claims?"
      │            "What documents do you have?"
      ▼
GOALS           ─ "Do you have destination preferences?"
      │            "Family or connections in specific countries?"
      │            "Languages spoken?"
      ▼
REVIEW          ─ Structured summary, user confirms
      │
      ▼
ASSESSMENT      ← Phase 3 begins
      │
      ▼
RECOMMENDATIONS ← Phase 4 begins
      │
      ▼
DOCUMENTS       ← Phase 5 begins
      │
      ▼
COMPLETE
```

**Session object schema:**

```python
{
  "session_id": str,           # UUID
  "language": str,             # ISO 639-1 (e.g. "fr", "ar", "sw")
  "state": str,                # current state machine state
  "interview": {
    "origin_country": str | None,
    "current_country": str | None,
    "persecution_types": list[str],   # ["political", "ethnic", ...]
    "immediate_danger": bool | None,
    "family_situation": str | None,
    "documents_available": list[str],
    "languages_spoken": list[str],
    "destination_preferences": list[str],
    "prior_claims": bool | None,
    "displacement_duration": str | None,
    "free_text_history": str | None,
  },
  "assessment": {
    "convention_grounds": list[str],  # 1951 + AU Refugee Convention
    "risk_level": str,                # "high" | "moderate" | "low"
    "reasoning_trace": str,           # full visible reasoning text
    "recommended_countries": list[CountryRecommendation],
  },
  "selected_country": str | None,
  "messages": list[dict],            # full conversation history
  "created_at": str,                 # ISO 8601
  "updated_at": str,
}
```

---

### 4. Gradio Phases (`app/phases/`)

Each phase is a self-contained Python module that exports a `build(session)`
function returning a `gr.Blocks` or `gr.Column` component tree.

**Phase ↔ State contract:**

| Phase module | Entry state | Exit state | Gradio yield pattern |
|---|---|---|---|
| `intake.py` | `LANGUAGE_SELECT` | `INTAKE` | `gr.update()` on language pill click |
| `interview.py` | `INTAKE` | `REVIEW` | Streaming via `AgentLoop.run()` async generator |
| `assessment.py` | `ASSESSMENT` | `RECOMMENDATIONS` | Streaming text delta to `gr.Textbox` |
| `recommendations.py` | `RECOMMENDATIONS` | `DOCUMENTS` | Card selection + roadmap render |
| `documents.py` | `DOCUMENTS` | `COMPLETE` | PDF download via `gr.File` |

**Consuming the agent loop in Gradio:**

```python
# In interview.py — streaming loop events into Gradio
async def on_submit(user_input, session_state):
    async for event in agent_loop.run(user_input, session_state, ...):
        if isinstance(event, TextDeltaEvent):
            yield gr.update(value=accumulated_text + event.delta)
        elif isinstance(event, ToolStartEvent):
            yield gr.update(value=f"[searching: {event.name}...]")
        elif isinstance(event, AgentEndEvent):
            session_state.messages = event.messages
            yield gr.update(value=accumulated_text)
```

**Gradio theming contract:**

All Gradio components use `elem_id` and `elem_classes` so custom CSS (injected
via `gr.HTML` or `gr.Blocks(css=...)`) can target them with DESIGN.md tokens.

Example mapping:
```python
gr.Button("Begin", elem_classes=["btn-primary"])  # → amber, DESIGN.md button-primary
gr.Button("Save and continue later", elem_classes=["btn-ghost"])
```

The CSS variable block from `DESIGN.md` tokens is injected as a `:root` block
at app startup. Never hardcode hex values in Python.

---

### 5. Document Generator (`agent/tools/doc_generator.py`)

Generates the Phase 5 document package. Input: completed session object.
Output: list of file paths (PDF + text).

**Documents produced:**

| File | Content | Pre-filled from |
|---|---|---|
| `personal_statement.pdf` | Structured personal statement template | `session.interview.*` |
| `action_plan.pdf` | Country-specific step-by-step action plan | `session.selected_country` + assessment |
| `emergency_contacts.pdf` | UNHCR offices + legal aid orgs | Country lookup tool result |
| `rights_summary_card.pdf` | Plain-language rights card | Selected country asylum law |

**Pre-filling rules:**
- Every pre-filled field is tagged with a visual amber highlight in the PDF
  (per `DESIGN.md §7 document-item`)
- No field is pre-filled with data not present in `session.interview`
- The document generator MUST log every field it fills and its source key

---

## Data Flow: Full Interview → Document

```
User selects language
        │
        ▼
Gradio intake.py → session.language = "fr"
        │
        ▼
agent_loop.run("Begin interview in French", session, ...)
        │  turns: SITUATION → HISTORY → GOALS → REVIEW
        │  tools: none in this phase
        │
        ▼
Events stream back → interview.py renders chat bubbles + structured responders
        │
        ▼
session.state = REVIEW → user confirms summary
        │
        ▼
agent_loop.run("Assess this case", session, thinking_level="medium")
        │
        ▼
  tool: web_search("Ethiopia asylum seeker Sudan UNHCR 2026")
  tool: country_lookup("Kenya", profile={origin:"Ethiopia", persecution:"political"})
  tool: country_lookup("Uganda", ...)
        │
        ▼
Events stream → assessment.py renders reasoning trace progressively
        │
        ▼
AgentEndEvent carries structured assessment JSON in final message
        │
        ▼
session.assessment populated → recommendations.py renders country cards
        │
        ▼
User selects country → session.selected_country = "Kenya"
        │
        ▼
doc_generator.py generates 4 PDFs from session
        │
        ▼
documents.py renders download list + preview
```

---

## Performance Targets

| Metric | Target | Notes |
|---|---|---|
| First agent token latency | < 3s | From user submission to first streaming character |
| Interview turn round-trip | < 8s | Single question turn on 7B model, Ollama local |
| Assessment completion | < 45s | Includes 2–3 tool calls + full reasoning trace |
| Document generation | < 10s | All 4 PDFs |
| App cold start | < 8s | Gradio only — no subprocess spawn needed |

---

## Security and Privacy Constraints

1. **No external logging of personal data.** Session objects are in-memory only
   for the hackathon. No database writes, no external analytics calls.
2. **Tool calls for country data only.** The `web_search` tool MUST be scoped
   to asylum/UNHCR/safety queries. Queries containing personal details from
   the interview MUST NOT be sent to external search APIs.
3. **Document files are ephemeral.** Generated PDFs are stored in a temp
   directory and deleted after download or session end.
4. **No PII in logs.** Log only state transitions and tool call metadata
   (not content). Never log interview answers.

---

## Dependency Versions (pinned)

```
# Python only — no Node.js dependencies
gradio==4.x
weasyprint==62.x
httpx==0.27.x
pytest==8.x
ollama==0.x          # LLM calls to local Ollama instance
litellm==1.x         # Optional: multi-provider abstraction (Ollama + Modal + Anthropic)
```

Lock file (`requirements.txt`) is committed.
Do not upgrade dependencies during a phase without explicit developer approval.