diff --git a/crates/mem-cli/src/agent/client_sdk.rs b/crates/mem-cli/src/agent/client_sdk.rs index b204867..e4262c7 100644 --- a/crates/mem-cli/src/agent/client_sdk.rs +++ b/crates/mem-cli/src/agent/client_sdk.rs @@ -249,6 +249,44 @@ impl SynthesisClient { self.execute_with_operation(req, Some("synthesis/infer")) .await } + + /// Execute Temporal workflow via api.riotpiao.com/workflow + /// Calls gateway which handles gRPC to Temporal service + pub async fn execute_workflow( + &self, + workflow_request: serde_json::Value, + ) -> Result { + let client = reqwest::Client::new(); + let url = format!("{}/workflow", self.base_url); + let start_time = std::time::Instant::now(); + + let response = client + .post(&url) + .header("Authorization", format!("Bearer {}", self.jwt_token)) + .header("Content-Type", "application/json") + .json(&workflow_request) + .timeout(std::time::Duration::from_secs(self.timeout_secs as u64)) + .send() + .await + .map_err(|e| format!("Workflow request failed: {}", e))?; + + let elapsed_ms = start_time.elapsed().as_millis() as u32; + + if response.status().is_success() { + let body = response + .json::() + .await + .map_err(|e| format!("Failed to parse workflow response: {}", e))?; + tracing::debug!("Workflow executed in {}ms", elapsed_ms); + Ok(body) + } else { + let error_text = response + .text() + .await + .unwrap_or_else(|_| "unknown error".to_string()); + Err(format!("Workflow failed ({}): {}", response.status(), error_text)) + } + } } #[cfg(test)] diff --git a/crates/mem-cli/src/handlers/agent_handler.rs b/crates/mem-cli/src/handlers/agent_handler.rs index f045be1..971168b 100644 --- a/crates/mem-cli/src/handlers/agent_handler.rs +++ b/crates/mem-cli/src/handlers/agent_handler.rs @@ -104,6 +104,35 @@ pub async fn register_agent_handler( info!("Agent registered: {}", agent.config().agent_id); + // Wire Temporal workflow (via api.riotpiao.com/workflow) + if let Some(jwt) = extract_jwt_token(&req) { + let client = SynthesisClient::new( + "https://api.riotpiao.com".to_string(), + jwt, + ); + + // Start Temporal workflow for agent initialization + let workflow_req = serde_json::json!({ + "action": "START_WORKFLOW", + "namespace": "poimen", + "payload": { + "workflow_id": format!("agent-init-{}", body.agent_id), + "workflow_type": "AgentInitialization", + "task_queue": "agents", + "input": { + "agent_id": body.agent_id, + "capabilities": body.capabilities, + "project_id": body.project_id + } + } + }); + + // Note: Call would be: + // let resp = client.execute_workflow(workflow_req).await; + // Store workflow_id/run_id in temporal_workflow_links table + debug!("Workflow request prepared (TODO: execute via SynthesisClient)"); + } + response_builder::success_response(AgentResponse { agent_id: agent.config().agent_id.clone(), project_id: agent.config().project_id.clone(), diff --git a/docs/TEMPORAL_WORKFLOW_INTEGRATION.md b/docs/TEMPORAL_WORKFLOW_INTEGRATION.md new file mode 100644 index 0000000..d72f53a --- /dev/null +++ b/docs/TEMPORAL_WORKFLOW_INTEGRATION.md @@ -0,0 +1,293 @@ +# 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` + +```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 Lifecycle + +### 1. Register Agent +``` +POST /agents + → START_WORKFLOW (AgentInitialization) + → Store workflow_id, run_id + → Return AgentResponse +``` + +### 2. Monitor Execution +``` +GET /agents/{id} + → DESCRIBE_WORKFLOW (query Temporal) + → Return execution status +``` + +### 3. Agent Reasoning Calls +``` +POST /synthesis/reason + → Temporal workflow sends events to agent + → Agent uses SynthesisClient.reason_query() to call LLM + → Store reasoning traces +``` + +### 4. Cleanup +``` +DELETE /agents/{id} + → 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 { + 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). diff --git a/k8s/config/temporal-endpoints.enc.yaml b/k8s/config/temporal-endpoints.enc.yaml new file mode 100644 index 0000000..b78fbb3 Binary files /dev/null and b/k8s/config/temporal-endpoints.enc.yaml differ