poimen-workflows/SERVICE_INTEGRATION.md

Poimen Service Integration & Declarative Workflow Design

Overview

Current (Bad): Hardcoded workflows

CLI flags → OrchestratorWorkflow (fixed flow)

Proposed (Better): API-driven, customer-specified workflows

API Request (JSON) → Routing Agent → Dynamic Activity Sequence → Results

---

Part 1: Service Integration Patterns (AWS Step Functions Inspired)

Pattern 1: Sequential (Request-Response)

# Customer specifies: "Call A, then call B with A's output"
activities:
  - name: CloneRepo
    type: request-response
    resource: CloneRepoActivity
    parameters:
      repo: "https://github.com/..."
      path: "/tmp/clone"
    timeout: 5m
    retry:
      maxAttempts: 3
      backoff: exponential
  
  - name: Analyze
    type: request-response
    resource: AnalyzeCodeActivity
    parameters:
      path: $.CloneRepo.path  # ← Reference previous output
    inputPath: "$.CloneRepo"   # ← What to pass
    outputPath: "$"            # ← What to return
    timeout: 10m

Execution: CloneRepo waits → outputs result → Analyze receives it → outputs

---

Pattern 2: Await-Task-Complete (Callback)

# Customer specifies: "Call A, wait for callback, then call B"
activities:
  - name: LaunchJob
    type: await-callback
    resource: LaunchLongRunningJobActivity
    parameters:
      job_id: "job-123"
    heartbeat_timeout: 5m
    next: OnJobComplete
  
  - name: OnJobComplete
    type: request-response
    resource: ProcessResultActivity
    parameters:
      result: $.LaunchJob.result

Execution: LaunchJob sent → waits for callback (heartbeat) → ProcessResult receives callback data → outputs

---

Pattern 3: Retry with Backoff

# Customer specifies: "Call A, if fails retry 3x with exponential backoff, else B"
activities:
  - name: GenerateCode
    type: request-response
    resource: ImplementerActivity
    parameters:
      task: $.input.taskID
      model: $.input.model
    retry:
      maxAttempts: 3
      backoffRate: 2.0        # 1s → 2s → 4s
      initialInterval: 1s
      maxInterval: 10m
    catch:
      - errorEquals: ["Timeout", "ThrottlingException"]
        next: HandleError
      - errorEquals: ["States.ALL"]
        next: HandleError
    next: JudgeCode
  
  - name: JudgeCode
    type: request-response
    resource: JudgeActivity
    parameters:
      implementation: $.GenerateCode.code
  
  - name: HandleError
    type: pass
    result: { status: "failed", reason: $.error }
    resultPath: "$.error_output"
    end: true

Execution: GenerateCode fails → retry 1 (wait 1s) → fails → retry 2 (wait 2s) → fails → HandleError

---

Part 2: Routing Agent Architecture

2.1 Routing Agent Activity (New)

// internal/routing/routing_agent.go

type WorkflowSpec struct {
    WorkflowID    string           `json:"workflow_id"`
    Name          string           `json:"name"`
    Input         map[string]interface{} `json:"input"`
    Activities    []ActivityStep   `json:"activities"`
    RetryPolicy   RetryConfig      `json:"retry_policy,omitempty"`
}

type ActivityStep struct {
    Name          string                 `json:"name"`
    Type          string                 `json:"type"` // request-response, await-callback, parallel, choice
    Resource      string                 `json:"resource"` // Activity name
    Parameters    map[string]interface{} `json:"parameters"`
    InputPath     string                 `json:"input_path"` // JSONPath: "$" or "$.previous.output"
    OutputPath    string                 `json:"output_path"` // What to extract
    Timeout       time.Duration          `json:"timeout"`
    Retry         *RetryConfig           `json:"retry,omitempty"`
    Catch         []CatchBlock           `json:"catch,omitempty"`
    Next          string                 `json:"next,omitempty"`
    End           bool                   `json:"end,omitempty"`
}

type RetryConfig struct {
    MaxAttempts      int           `json:"max_attempts"`
    BackoffRate      float64       `json:"backoff_rate"`
    InitialInterval  time.Duration `json:"initial_interval"`
    MaxInterval      time.Duration `json:"max_interval"`
}

type CatchBlock struct {
    ErrorEquals []string `json:"error_equals"`
    Next        string   `json:"next"`
    ResultPath  string   `json:"result_path,omitempty"`
}

// RoutingAgentActivity: Receives WorkflowSpec, executes activities sequentially
func RoutingAgentActivity(ctx context.Context, spec WorkflowSpec) (map[string]interface{}, error) {
    // 1. Parse WorkflowSpec
    // 2. Build execution plan (DAG of activities)
    // 3. For each step:
    //    a. Evaluate JSONPath inputs
    //    b. Execute activity with retries
    //    c. Handle errors (catch blocks)
    //    d. Store output in context
    //    e. Move to next step
    // 4. Return final result
}

2.2 Activity Mapper Registry (New)

// internal/routing/activity_registry.go

type ActivityMapper struct {
    activities map[string]interface{}
}

func NewActivityMapper() *ActivityMapper {
    return &ActivityMapper{
        activities: map[string]interface{}{
            "CloneRepoActivity":      action.CloneRepoActivity,
            "AnalyzeCodeActivity":    action.AnalyzeCodeActivity,
            "ImplementerActivity":    action.ImplementerActivity,
            "JudgeActivity":          action.JudgeActivity,
            "GitCommitActivity":      action.GitCommitActivity,
            // ... all activities registered here
        },
    }
}

func (am *ActivityMapper) Get(name string) (interface{}, error) {
    if activity, ok := am.activities[name]; ok {
        return activity, nil
    }
    return nil, fmt.Errorf("activity %s not found", name)
}

// cmd/worker/main.go
w.RegisterActivity(routing.RoutingAgentActivity)  // ← Only register this
w.RegisterActivity(action.CloneRepoActivity)      // ← Still register all individual activities
// ... rest of activities

---

Part 3: API Message Format (General Purpose)

3.1 Workflow Request Format

POST /api/workflows
{
  "workflow": {
    "id": "workflow-123",
    "name": "code-review-pipeline",
    "version": "v1",
    "input": {
      "repo": "https://github.com/rockliang/poimen",
      "branch": "feature/new-feature",
      "milestone": "T0"
    },
    "activities": [
      {
        "id": "step_1",
        "name": "CloneRepo",
        "type": "request-response",
        "resource": "CloneRepoActivity",
        "parameters": {
          "repo": "${input.repo}",
          "path": "/tmp/work"
        },
        "timeout": "5m",
        "next": "step_2"
      },
      {
        "id": "step_2",
        "name": "AnalyzeCode",
        "type": "request-response",
        "resource": "AnalyzeCodeActivity",
        "parameters": {
          "path": "${step_1.output.path}"
        },
        "inputPath": "$.step_1",
        "outputPath": "$",
        "timeout": "10m",
        "retry": {
          "maxAttempts": 3,
          "backoffRate": 2.0,
          "initialInterval": "1s"
        },
        "catch": [
          {
            "errorEquals": ["Timeout"],
            "next": "handle_timeout"
          }
        ],
        "next": "step_3"
      },
      {
        "id": "step_3",
        "name": "GenerateImplementation",
        "type": "request-response",
        "resource": "ImplementerActivity",
        "parameters": {
          "analysis": "${step_2.output.analysis}",
          "model": "claude-sonnet-5"
        },
        "timeout": "30m",
        "retry": {
          "maxAttempts": 3,
          "backoffRate": 2.0
        },
        "next": "step_4"
      },
      {
        "id": "step_4",
        "name": "Review",
        "type": "request-response",
        "resource": "JudgeActivity",
        "parameters": {
          "code": "${step_3.output.code}",
          "analysis": "${step_2.output.analysis}"
        },
        "timeout": "10m",
        "end": true
      },
      {
        "id": "handle_timeout",
        "name": "TimeoutHandler",
        "type": "pass",
        "result": {
          "status": "failed",
          "reason": "Analysis timed out"
        },
        "end": true
      }
    ]
  }
}

3.2 Workflow Response Format

{
  "workflow_id": "workflow-123",
  "status": "RUNNING",
  "temporal_workflow_id": "orch-workflow-123",
  "task_queue": "poimen-taskqueue",
  "execution_started_at": "2025-01-29T10:00:00Z",
  "monitoring": {
    "web_ui": "http://temporal.local:8080/namespaces/poimen/workflows/orch-workflow-123",
    "poll_url": "/api/workflows/workflow-123/status"
  }
}

3.3 Status Poll Format

GET /api/workflows/workflow-123/status
{
  "workflow_id": "workflow-123",
  "status": "COMPLETED",
  "steps_completed": 4,
  "total_steps": 4,
  "results": {
    "step_1": {
      "name": "CloneRepo",
      "status": "COMPLETED",
      "duration": "45s",
      "output": {
        "path": "/tmp/work/poimen",
        "commit": "abc123"
      }
    },
    "step_2": {
      "name": "AnalyzeCode",
      "status": "COMPLETED",
      "duration": "120s",
      "output": {
        "analysis": "..." 
      }
    },
    "step_3": {
      "name": "GenerateImplementation",
      "status": "COMPLETED",
      "duration": "300s",
      "output": {
        "code": "..."
      }
    },
    "step_4": {
      "name": "Review",
      "status": "COMPLETED",
      "duration": "60s",
      "output": {
        "verdict": "approved",
        "score": 0.95
      }
    }
  },
  "final_output": {
    "verdict": "approved",
    "score": 0.95
  },
  "completed_at": "2025-01-29T10:10:00Z"
}

---

Part 4: Implementation Plan

Phase 1: Core Routing Engine (Week 1)

• Create internal/routing/types.go (WorkflowSpec, ActivityStep structs)
• Create internal/routing/jsonpath_evaluator.go (parameter interpolation)
• Create internal/routing/execution_context.go (execution state management)
• Create internal/routing/activity_registry.go (activity mapper)
• Create statemachine/routing_workflow.go (RoutingWorkflow implementation)
• Update cmd/worker/main.go (register RoutingWorkflow)
• Write tests for routing engine

Phase 2: API Server (Week 2)

• Create cmd/api-server/main.go (HTTP server bootstrap)
• Create cmd/api-server/handlers.go (POST /workflows, GET /status)
• Create cmd/api-server/validation.go (WorkflowSpec validation)
• Create internal/routing/client.go (Temporal client wrapper)
• Write integration tests

Phase 3: Homelab Frontend Integration (Week 3)

• Create k8s/gateway-routes.yaml (route /api/v1/workflows to api-server)
• Deploy api-server as K8s Service
• Update NetworkPolicy to allow traffic to api-server
• Test end-to-end API flow

Phase 4: Migration & Cleanup (Week 4)

• Keep cmd/starter for backward compatibility
• Document migration path
• Update examples to use new API
• Performance tuning

---

Files to Create

NEW:
  internal/routing/
    ├─ types.go                    (WorkflowSpec, ActivityStep, RetryConfig)
    ├─ jsonpath_evaluator.go       (Parameter interpolation)
    ├─ execution_context.go        (State management)
    ├─ activity_registry.go        (Activity mapper)
    └─ routing_workflow.go         (RoutingWorkflow implementation)
  
  statemachine/
    └─ routing_workflow.go         (See Part 5 below)
  
  cmd/api-server/
    ├─ main.go                     (Server bootstrap)
    ├─ handlers.go                 (HTTP handlers)
    └─ validation.go               (Spec validation)

MODIFY:
  cmd/worker/main.go              (Register RoutingWorkflow)
  
NEW (Homelab Frontend):
  k8s/
    └─ gateway-routes.yaml         (Route definitions)

---

Backward Compatibility

✅ Existing cmd/starter still works (calls OrchestratorWorkflow) ✅ New API calls RoutingWorkflow ✅ Worker registers both workflows ✅ No breaking changes to current deployments

---

Benefits

✅ Customer-driven workflows (JSON API) ✅ Service Integration patterns (Sequential, Await-Task-Complete, Retry) ✅ No recompilation needed ✅ Full Temporal observability ✅ JSONPath-based parameter passing ✅ Composable activities ✅ Error handling & retries built-in