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>

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

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.)

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:

   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:

   let client = SynthesisClient::new(url, jwt_token);
   

4. Client includes token in all requests:

   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):

# 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:

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

{
  "success": false,
  "error": "WORKFLOW_NOT_FOUND",
  "message": "workflow_id 'agent-init-xyz' not found in namespace 'poimen'"
}

Temporal Unreachable

{
  "success": false,
  "error": "TEMPORAL_UNAVAILABLE",
  "message": "Temporal server connection not available"
}

Invalid JWT

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

Testing

Mock Workflow Response

#[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).