poimen-memory/docs/TEMPORAL_WORKFLOW_INTEGRATION.md

# Temporal Workflow Service Integration

## Architecture

Memory service delegates **agent execution** to Temporal.io (external service), which runs long-lived workflows. Memory DB only stores **references + reasoning traces**.

```
┌─────────────────────────────────────────────────┐
│ POST /agents (register agent)                   │
│  ↓                                              │
│  Extract JWT from Authorization header         │
│  Create SynthesisClient with JWT               │
│  Call: POST api.riotpiao.com/workflow          │
│  Body: {                                        │
│    "action": "START_WORKFLOW",                 │
│    "namespace": "poimen",                       │
│    "payload": {                                 │
│      "workflow_id": "agent-init-{agent_id}",   │
│      "workflow_type": "AgentInitialization",  │
│      "task_queue": "agents",                    │
│      "input": {...}                             │
│    }                                            │
│  }                                              │
│  ↓                                              │
│  Receive: { workflow_id, run_id, status }      │
│  Store in: temporal_workflow_links table        │
│  ↓                                              │
│  Return AgentResponse to client                 │
└─────────────────────────────────────────────────┘
```

## API Gateway (api.riotpiao.com)

Gateway translates REST to Temporal gRPC:

**Endpoint:** `POST /workflow`

**Supported Actions:**
- `START_WORKFLOW` — Begin new execution
- `DESCRIBE_WORKFLOW` — Get status
- `CANCEL_WORKFLOW` — Request graceful stop
- `SIGNAL_WORKFLOW` — Send signal to running workflow
- `GET_WORKFLOW_HISTORY` — Retrieve event log

See: `/Users/rockliang/workplace/homelab-frontend/API.md#workflow-services-temporal`

## Code Integration Points

### 1. SynthesisClient Extension

File: `crates/mem-cli/src/agent/client_sdk.rs`

New method: `execute_workflow(workflow_request: Value) -> Result<Value, String>`

```rust
let workflow_req = serde_json::json!({
    "action": "START_WORKFLOW",
    "namespace": "poimen",
    "payload": {
        "workflow_id": "agent-init-{agent_id}",
        "workflow_type": "AgentInitialization",
        "task_queue": "agents",
        "input": { "agent_id": agent_id, ... }
    }
});

let response = client.execute_workflow(workflow_req).await?;
let run_id = response["data"]["run_id"].as_str().unwrap();
```

### 2. Agent Handler Wiring

File: `crates/mem-cli/src/handlers/agent_handler.rs`

```rust
pub async fn register_agent_handler(...) {
    // Extract JWT token
    let jwt = extract_jwt_token(&req)?;
    
    // Create client with JWT
    let client = SynthesisClient::new(
        "https://api.riotpiao.com".to_string(),
        jwt,
    );
    
    // Call workflow via gateway
    let workflow_req = serde_json::json!({...});
    let resp = client.execute_workflow(workflow_req).await?;
    
    // Store workflow_id/run_id reference
    let workflow_id = resp["data"]["workflow_id"].as_str().unwrap();
    let run_id = resp["data"]["run_id"].as_str().unwrap();
    
    // Insert into temporal_workflow_links table
    db.create_workflow_link(workflow_id, run_id, None, None, None).await?;
}
```

### 3. Database Schema

File: `crates/mem-store/migrations/005_workflows_schema.sql`

Table: `temporal_workflow_links`
- `id` — PK (UUID)
- `workflow_id` — Temporal workflow ID (external reference)
- `run_id` — Temporal run ID (external reference)
- `entity_id`, `edge_id`, `node_id` — Optional Memory DB references
- `status` — active|completed|failed|archived
- `metadata` — Extra context (error details, etc.)

```sql
INSERT INTO temporal_workflow_links 
  (workflow_id, run_id, status, created_at)
VALUES 
  ('agent-init-agent1', 'abc123', 'active', NOW());
```

## Workflow & Activity Pattern

**Key Insight**: Memory service orchestrates via REST, Temporal activities do the actual LLM work + persistence.

```
Memory Handler                      Temporal Workflow              LLM API
    |
    v
Extract JWT token
    |
    v
SynthesisClient.execute_workflow(
  {"action": "START_WORKFLOW", ...}
)  ─────────────────────────────→  ReasoningWorkflow
                                      |
                                      v
                                  InferenceActivity
                                      |
                                      v (LLM call)
                                   api.riotpiao.com/v1/chat/completions
                                      |
                                      v (with JWT)
                                   LLM Response
                                      |
                                      v (persist)
                                   memory_entity + memory_edge
                                      |
                                      v (activity completes)
                                  Workflow completes
                                      |
Poll DESCRIBE_WORKFLOW ←───────────  Return result
    |
    v
Memory handler receives result
    |
    v
Return to client
```

## Workflow Lifecycle

### 1. Register Agent
```
POST /agents
  → Extract JWT from Authorization header
  → SynthesisClient.execute_workflow({"action": "START_WORKFLOW", ...})
    └─ Gateway: POST /workflow + JWT header
    └─ Temporal: Start AgentInitialization workflow
      └─ Activity: Persist to temporal_workflow_links table
  → Store workflow_id, run_id
  → Return AgentResponse to client
```

### 2. Reasoning Workflow
```
POST /memory/synthesis
  → JWT required (Bearer token)
  → SynthesisClient.execute_workflow(ReasoningWorkflow)
    └─ Gateway: POST /workflow + JWT
    └─ Temporal: Start ReasoningWorkflow
      └─ Activity: InferenceActivity
        ├─ Call: POST api.riotpiao.com/v1/chat/completions (with JWT)
        ├─ Parse LLM response
        └─ Persist result to memory_entity/memory_edge (temporal_workflow_links tracks link)
  → Poll DESCRIBE_WORKFLOW (retry up to 30×, 100ms interval)
  → Wait for status COMPLETED or FAILED
  → Return ReasoningResult to client
```

### 3. Durability & Checkpointing
```
Activity Crash during LLM call
  → Temporal replays workflow from last checkpoint
  → Activity re-executes (idempotent via idempotency_key)
  → Result persisted to DB (already done from retry)
  → No data loss
```

### 4. Monitor Execution
```
GET /agents/{id}
  → Extract JWT
  → SynthesisClient.execute_workflow({"action": "DESCRIBE_WORKFLOW", ...})
  → Return execution status (RUNNING|COMPLETED|FAILED)
```

### 5. Cleanup
```
DELETE /agents/{id}
  → JWT required
  → CANCEL_WORKFLOW
  → Mark temporal_workflow_links.status = 'archived'
```

## JWT Auth Flow

**Per-Request Bearer Token:**

1. Client provides JWT in Authorization header
2. Agent handler extracts token:
   ```rust
   fn extract_jwt_token(req: &HttpRequest) -> Option<String> {
       req.headers()
           .get("Authorization")
           .and_then(|h| h.to_str().ok())
           .and_then(|s| {
               if s.starts_with("Bearer ") {
                   Some(s[7..].to_string())
               } else {
                   None
               }
           })
   }
   ```

3. Token passed to SynthesisClient:
   ```rust
   let client = SynthesisClient::new(url, jwt_token);
   ```

4. Client includes token in all requests:
   ```rust
   client
       .post(url)
       .header("Authorization", format!("Bearer {}", self.jwt_token))
       .send()
       .await
   ```

5. Gateway validates token via Authentik, forwards to Temporal

## Configuration

### Environment Variables (Pod-Aware)

Loaded from ConfigMap (encrypted with SOPS):

```yaml
# k8s/config/temporal-endpoints.enc.yaml
# (SOPS/age encrypted, auto-decrypted by ArgoCD+KSOPS)

INTERNAL_SYNTHESIS_URL: http://api-gw.riotpiao.svc.cluster.local:8080
EXTERNAL_SYNTHESIS_URL: https://api.riotpiao.com
SYNTHESIS_TIMEOUT_SECS: 30
```

Pod-aware detection:
```rust
let is_in_pod = env::var("KUBERNETES_SERVICE_HOST").is_ok();
let base_url = if is_in_pod {
    env::var("INTERNAL_SYNTHESIS_URL").unwrap_or(external)
} else {
    env::var("EXTERNAL_SYNTHESIS_URL").unwrap_or(external)
};
```

## Error Handling

### Workflow Not Found
```json
{
  "success": false,
  "error": "WORKFLOW_NOT_FOUND",
  "message": "workflow_id 'agent-init-xyz' not found in namespace 'poimen'"
}
```

### Temporal Unreachable
```json
{
  "success": false,
  "error": "TEMPORAL_UNAVAILABLE",
  "message": "Temporal server connection not available"
}
```

### Invalid JWT
```json
{
  "success": false,
  "error": "UNAUTHORIZED",
  "message": "Invalid or expired JWT token"
}
```

## Testing

### Mock Workflow Response

```rust
#[test]
async fn test_register_agent_with_workflow() {
    let jwt = "test-jwt-token";
    let client = SynthesisClient::new(
        "https://api.riotpiao.com".to_string(),
        jwt.to_string(),
    );
    
    // Mock: POST /workflow returns
    let workflow_response = serde_json::json!({
        "success": true,
        "data": {
            "workflow_id": "agent-init-test",
            "run_id": "abc123",
            "status": "RUNNING"
        }
    });
    
    // Assert workflow_id captured
    assert_eq!(workflow_response["data"]["workflow_id"], "agent-init-test");
}
```

## TODO: Production Implementation

1. **Agent DB Persistence**
   - Save agents to PostgreSQL `agents` table
   - Instead of in-memory DefaultAgent struct

2. **Workflow Link Storage**
   - Actually call `temporal_workflow_links` insert
   - Currently stub in agent_handler.rs

3. **Status Queries**
   - Implement GET /agents/{id}
   - Call DESCRIBE_WORKFLOW to poll Temporal

4. **Webhook Callbacks**
   - Temporal sends workflow events to webhook_url
   - Memory service processes events (entity updates, etc.)

5. **Integration Tests**
   - Mock api.riotpiao.com responses
   - End-to-end: register → workflow → query

---

**Status**: Architecture wired, implementation stubbed. Ready for Phase 6.5 (production hardening).