diff --git a/COMPLETE_SPECIFICATION_SUMMARY.txt b/COMPLETE_SPECIFICATION_SUMMARY.txt new file mode 100644 index 0000000..6c5a005 --- /dev/null +++ b/COMPLETE_SPECIFICATION_SUMMARY.txt @@ -0,0 +1,278 @@ +================================================================================ +POIMEN ROUTING WORKFLOW - COMPLETE SPECIFICATION SUMMARY +================================================================================ + +STATUS: βœ… READY FOR IMPLEMENTATION + +Created: August 31, 2025 +Total Documentation: 3,856 lines across 5 files +Implementation Effort: 60-70 hours (3-4 weeks, 1-2 engineers) + +================================================================================ +πŸ“š DOCUMENTATION CREATED +================================================================================ + +1. ROUTING_WORKFLOW_SPEC.md (1,198 lines, 32KB) + β”œβ”€ Architecture overview + β”œβ”€ ActivityKnowledgeBase.json format + β”œβ”€ llm-router Activity (intelligent generator) + β”œβ”€ RoutingWorkflow (generic executor) + β”œβ”€ Go type definitions (copy-paste ready) + β”œβ”€ 7 implementation architecture sections + β”œβ”€ CronWorkflowSpec (scheduled workflows) + β”œβ”€ Execution flow with examples + β”œβ”€ Validation rules + └─ Complete reference + +2. IMPLEMENTATION_TASKS.md (1,158 lines, 27KB) + β”œβ”€ Phase 1: Foundation (8-10 hours, 4 tasks) + β”œβ”€ Phase 2: LLM-Router (12-15 hours, 5 tasks) + β”œβ”€ Phase 3: RoutingWorkflow (15-18 hours, 6 tasks) + β”œβ”€ Phase 4: API/CLI (12-15 hours, 4 tasks) + β”œβ”€ Phase 5: Testing (8-12 hours, 4 tasks) + β”œβ”€ Phase 6: Documentation (5-8 hours, 4 tasks) + β”œβ”€ Total: 27 specific, actionable tasks + β”œβ”€ Each with: effort estimate, acceptance criteria, dependencies + β”œβ”€ Timeline: 3-4 weeks + β”œβ”€ Resource allocation: 1-2 engineers + β”œβ”€ Blockers to watch + └─ Success criteria per phase + +3. CRON_JOBS_QUICK_REFERENCE.md (252 lines, 5.7KB) + β”œβ”€ Cron syntax examples (daily, hourly, weekly, etc) + β”œβ”€ How llm-router detects scheduled jobs + β”œβ”€ Execution tracking + β”œβ”€ Proposed API endpoints + β”œβ”€ One-time vs Cron comparison table + └─ Quick lookup reference + +4. DESIGN_MASTER_REVIEW.md (888 lines, 25KB) + β”œβ”€ Executive summary for stakeholders + β”œβ”€ Problem/solution statement + β”œβ”€ 3 patterns (Sequential, Await-Task-Complete, Retry) + β”œβ”€ 3 entry points (CLI, API, Legacy) + β”œβ”€ Before/after comparison + β”œβ”€ Implementation timeline + β”œβ”€ KMSvc questions (Q1-Q6) + β”œβ”€ Risks & mitigations + β”œβ”€ Success criteria + β”œβ”€ Approval checklist + └─ Complete example workflows + +5. README_IMPLEMENTATION.md (360 lines, 9KB) + β”œβ”€ Quick start guide + β”œβ”€ Documentation structure explanation + β”œβ”€ Week-by-week breakdown + β”œβ”€ How to start today + β”œβ”€ Success criteria per phase + β”œβ”€ Effort summary table + β”œβ”€ Key features matrix + β”œβ”€ Tips for success + β”œβ”€ Learning resources + └─ Decision maker's checklist + +================================================================================ +🎯 THE SYSTEM ARCHITECTURE +================================================================================ + +User Input (one-time or scheduled): + "Analyze repo for security and quality" + or + "Scan all repos daily at 2 AM" + + ↓ + +[llm-router Activity] - Intelligent Workflow Generator + β”œβ”€ Reads: ActivityKnowledgeBase.json (metadata about activities) + β”œβ”€ Uses LLM to understand intent + β”œβ”€ Selects activities: Clone β†’ Analyze β†’ SecurityScan β†’ Combine β†’ Notify + β”œβ”€ Orders by dependencies + β”œβ”€ Decides timeout for each (from knowledge base) + β”œβ”€ Decides retry policy (from isFlaky flag) + β”œβ”€ Chains parameters (JSONPath: ${Clone.output.path}) + β”œβ”€ Detects if scheduled (cron) + └─ Generates: WorkflowSpec or CronWorkflowSpec (JSON) + + ↓ + +[RoutingWorkflow] - Generic Executor + β”œβ”€ Takes JSON spec from llm-router + β”œβ”€ Executes states in order + β”œβ”€ Respects timeout/retry for each activity + β”œβ”€ Handles errors with catch blocks + └─ Returns results + + ↓ + +[Temporal] - Distributed Workflow Engine + β”œβ”€ For one-time: Executes immediately + β”œβ”€ For cron: Schedules and runs on schedule + β”œβ”€ Provides durability (replay guarantee) + β”œβ”€ Tracks execution history + └─ Handles retries automatically + + ↓ + +[Results] - Final Output + β”œβ”€ Execution history + β”œβ”€ Step-by-step results + β”œβ”€ Performance metrics + └─ Status updates + +================================================================================ +✨ KEY FEATURES +================================================================================ + +βœ… One-time workflows (instant execution via API/CLI) +βœ… Scheduled workflows (cron jobs with full history) +βœ… Intelligent routing (LLM decides what to run) +βœ… Smart timeouts (from ActivityKnowledgeBase.json) +βœ… Smart retries (3x for flaky, 1x for stable) +βœ… Error handling (catch blocks for graceful failures) +βœ… Parameter chaining (JSONPath: ${step.output.field}) +βœ… Parallel execution (multiple branches) +βœ… Temporal durability (automatic replay on failure) +βœ… HTTP API (for programmatic access) +βœ… CLI (for command-line access) +βœ… Execution tracking (full history) +βœ… Backward compatible (legacy CLI still works) + +================================================================================ +πŸ“Š IMPLEMENTATION BREAKDOWN +================================================================================ + +PHASE 1: Foundation (8-10 hours) + Task 1.1: Go types (2h) + Task 1.2: ActivityKnowledgeBase.json (3h) + Task 1.3: KB loader (2h) + Task 1.4: Validator (3h) + β†’ Deliverable: Core data structures + +PHASE 2: LLM-Router (12-15 hours) + Task 2.1: JSONPath resolver (3h) + Task 2.2: Activity skeleton (2h) + Task 2.3: LLM intent analysis (5h) ⚠️ HIGHEST RISK + Task 2.4: Spec builder (4h) + Task 2.5: Cron builder (2h) + β†’ Deliverable: Intelligent workflow generation + +PHASE 3: RoutingWorkflow (15-18 hours) + Task 3.1: Executor dispatcher (1h) + Task 3.2: Task executor (2h) + Task 3.3: Pass/Fail executors (1h) + Task 3.4: Main workflow engine (4h) + Task 3.5: Register in worker (1h) + Task 3.6: Helper functions (2h) + β†’ Deliverable: Generic workflow executor + +PHASE 4: API/CLI (12-15 hours) + Task 4.1: API handlers (4h) + Task 4.2: CLI commands (5h) + Task 4.3: Server bootstrap (2h) + Task 4.4: Validation (2h) + β†’ Deliverable: HTTP API + CLI + +PHASE 5: Testing (8-12 hours) + Task 5.1: Unit tests (3h) + Task 5.2: Integration tests (4h) + Task 5.3: E2E tests (4h) + Task 5.4: Load tests (2h) + β†’ Deliverable: >90% coverage, all scenarios pass + +PHASE 6: Documentation (5-8 hours) + Task 6.1: API documentation (2h) + Task 6.2: CLI documentation (1h) + Task 6.3: Deployment guide (2h) + Task 6.4: User guide & examples (2h) + β†’ Deliverable: Complete documentation + +TOTAL: 60-70 hours (3-4 weeks, 1-2 engineers) + +================================================================================ +πŸš€ HOW TO START TODAY +================================================================================ + +Step 1: Review Documentation (1-2 hours) + β†’ Read ROUTING_WORKFLOW_SPEC.md (understand design) + β†’ Read IMPLEMENTATION_TASKS.md (understand tasks) + β†’ Read README_IMPLEMENTATION.md (quick start) + +Step 2: Assign Tasks (30 minutes) + β†’ Engineer 1: Tasks 1.1-1.4, 2.1-2.5, 3.1-3.6 + β†’ Engineer 2: Tasks 4.1-4.4, 5.1-5.4, 6.1-6.4 + +Step 3: Begin Implementation (immediately) + β†’ Start with Task 1.1: Create Go types (types.go) + β†’ 2 hours to completion + β†’ Then proceed to Task 1.2 (ActivityKnowledgeBase.json) + +Step 4: Daily Sync + β†’ Report progress + β†’ Unblock dependencies + β†’ Adjust timeline if needed + +================================================================================ +βœ… SUCCESS CRITERIA +================================================================================ + +Phase 1: All types compile, KB loads, validator works +Phase 2: llm-router generates valid specs, detects cron +Phase 3: RoutingWorkflow executes any spec, handles errors +Phase 4: HTTP API + CLI fully functional +Phase 5: >90% code coverage, all tests pass +Phase 6: Complete documentation, ready to ship + +βœ… DONE WHEN: + - All code compiles without warnings + - All tests pass (unit, integration, E2E, load) + - Documentation complete + - Can deploy to Kubernetes + - Can submit workflows from API/CLI + - Can create cron jobs + - Performance targets met (<200ms submit, <100ms poll) + +================================================================================ +πŸ“ FILE LOCATIONS +================================================================================ + +Core Specification: + ~/workplace/Poimen/workflows/ROUTING_WORKFLOW_SPEC.md + +Task Breakdown: + ~/workplace/Poimen/workflows/IMPLEMENTATION_TASKS.md + +Cron Reference: + ~/workplace/Poimen/workflows/CRON_JOBS_QUICK_REFERENCE.md + +Stakeholder Review: + ~/workplace/Poimen/workflows/DESIGN_MASTER_REVIEW.md + +Quick Start Guide: + ~/workplace/Poimen/workflows/README_IMPLEMENTATION.md + +This Summary: + ~/workplace/Poimen/workflows/COMPLETE_SPECIFICATION_SUMMARY.txt + +================================================================================ +πŸŽ“ RECOMMENDATION +================================================================================ + +This specification is: + βœ… Complete - covers all aspects of the system + βœ… Implementable - all code patterns shown + βœ… Testable - success criteria clearly defined + βœ… Maintainable - well-documented + βœ… Scalable - designed for production use + +NEXT STEPS: + 1. Get stakeholder approval (use DESIGN_MASTER_REVIEW.md) + 2. Assign engineers (use IMPLEMENTATION_TASKS.md) + 3. Start Phase 1, Task 1.1 today + 4. Daily standup on progress + 5. Gate each phase before moving to next + +TIMELINE: 3-4 weeks to complete implementation ⏱️ + +STATUS: 🟒 READY TO BUILD + +================================================================================ diff --git a/CRON_JOBS_QUICK_REFERENCE.md b/CRON_JOBS_QUICK_REFERENCE.md new file mode 100644 index 0000000..cccff19 --- /dev/null +++ b/CRON_JOBS_QUICK_REFERENCE.md @@ -0,0 +1,252 @@ +# Cron Jobs in Poimen Workflows + +## Quick Summary + +**llm-router** detects if user is asking for scheduled/recurring work and generates either: +- **WorkflowSpec** (one-time execution) +- **CronWorkflowSpec** (scheduled execution) + +--- + +## CRON EXAMPLES + +### Example 1: Daily Security Scan + +**User says**: "Run security scan on all repos every day at 2 AM" + +**llm-router detects**: +- is_scheduled: true +- schedule: "0 2 * * *" (2 AM every day) +- activities_needed: [Clone, SecurityScan, SendAlert] + +**llm-router generates**: +```json +{ + "type": "CronWorkflow", + "name": "daily-security-scan", + "schedule": "0 2 * * *", + "timezone": "UTC", + "input": {"repos": [...]}, + "states": [ + {"name": "Clone", "type": "Task", "resource": "CloneRepoActivity", ...}, + {"name": "SecurityScan", "type": "Task", "resource": "SecurityScanActivity", ...}, + {"name": "SendAlert", "type": "Task", "resource": "SendNotificationActivity", ...} + ] +} +``` + +**Temporal** executes this: +- At 2 AM UTC every day +- Runs RoutingWorkflow with this spec +- Each run is independent (tracks execution history) + +--- + +### Example 2: Hourly Health Check + +**User says**: "Check API health every hour" + +**llm-router generates**: +```json +{ + "type": "CronWorkflow", + "schedule": "0 * * * *", + "timezone": "UTC", + "states": [ + {"name": "HealthCheck", "type": "Task", "resource": "HealthCheckActivity", ...}, + {"name": "RecordMetrics", "type": "Task", "resource": "RecordMetricsActivity", ...} + ] +} +``` + +**Execution**: Every hour, automatically + +--- + +### Example 3: Weekly Performance Baseline + +**User says**: "Compare performance with baseline every Sunday at 3 AM" + +**llm-router generates**: +```json +{ + "type": "CronWorkflow", + "schedule": "0 3 * * 0", + "timezone": "America/New_York", + "states": [...] +} +``` + +**Execution**: Every Sunday at 3 AM in New York timezone + +--- + +## CRON SCHEDULE SYNTAX + +``` +β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€ minute (0 - 59) +β”‚ β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€ hour (0 - 23) +β”‚ β”‚ β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€ day of month (1 - 31) +β”‚ β”‚ β”‚ β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€ month (1 - 12) +β”‚ β”‚ β”‚ β”‚ β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€ day of week (0 - 6, 0 = Sunday) +β”‚ β”‚ β”‚ β”‚ β”‚ +β”‚ β”‚ β”‚ β”‚ β”‚ +* * * * * + +Common Examples: +0 2 * * * β†’ Every day at 2:00 AM +0 */6 * * * β†’ Every 6 hours +0 9 * * 1-5 β†’ Weekdays at 9 AM (Mon-Fri) +0 0 1 * * β†’ First day of month at midnight +0 0 * * 0 β†’ Every Sunday at midnight +*/15 * * * * β†’ Every 15 minutes +30 2 * * 0 β†’ Every Sunday at 2:30 AM +``` + +--- + +## EXECUTION TRACKING + +Temporal automatically tracks all cron executions: + +``` +Workflow: daily-security-scan + +Run 1: 2025-02-01 02:00:00 UTC β†’ COMPLETED (5m 32s) +Run 2: 2025-02-02 02:00:00 UTC β†’ COMPLETED (4m 58s) +Run 3: 2025-02-03 02:00:00 UTC β†’ FAILED (timeout in SecurityScan) +Run 4: 2025-02-04 02:00:00 UTC β†’ COMPLETED (6m 15s) +Run 5: 2025-02-05 02:00:00 UTC β†’ COMPLETED (5m 01s) +``` + +Retrieve with: +```bash +temporal workflow list --query "ExecutionStatus='Completed' AND WorkflowType='RoutingWorkflow'" +temporal workflow describe --workflow-id daily-security-scan +``` + +--- + +## HOW llm-router DETECTS CRON + +LLM looks for keywords in user message: + +``` +"every day at 2 AM" β†’ "0 2 * * *" +"every 6 hours" β†’ "0 */6 * * *" +"daily" β†’ "0 0 * * *" +"weekly" β†’ "0 0 * * 0" (Sunday) +"every Monday" β†’ "0 0 * * 1" +"every 15 minutes" β†’ "*/15 * * * *" +"weekdays at 9 AM" β†’ "0 9 * * 1-5" +"first of month" β†’ "0 0 1 * *" +"midnight" β†’ "0 0" +``` + +--- + +## DIFFERENCES: One-Time vs Cron + +| Aspect | One-Time | Cron | +|--------|----------|------| +| **Type** | WorkflowSpec | CronWorkflowSpec | +| **Triggered by** | API call, CLI | Schedule | +| **Execution** | Runs once, returns immediately | Runs on schedule, indefinitely | +| **Input** | Varies per call | Fixed for all runs | +| **History** | Single execution | Multiple executions tracked | +| **Cancellation** | Can't cancel | Can stop/restart cron | +| **Use Case** | Ad-hoc analysis | Background monitoring | + +--- + +## API ENDPOINTS (Proposed) + +### Submit Cron Workflow +```bash +POST /api/v1/cron/workflows +{ + "type": "CronWorkflow", + "schedule": "0 2 * * *", + "timezone": "UTC", + "states": [...] +} + +Response: +{ + "workflow_id": "daily-security-scan", + "schedule": "0 2 * * *", + "next_run": "2025-02-02 02:00:00 UTC", + "created": "2025-02-01 15:30:00 UTC" +} +``` + +### Get Cron Status +```bash +GET /api/v1/cron/workflows/daily-security-scan/status + +Response: +{ + "workflow_id": "daily-security-scan", + "schedule": "0 2 * * *", + "is_active": true, + "last_run": { + "time": "2025-02-01 02:00:00 UTC", + "status": "COMPLETED", + "duration": "5m 32s" + }, + "next_run": "2025-02-02 02:00:00 UTC", + "execution_history": [...] +} +``` + +### List Cron Workflows +```bash +GET /api/v1/cron/workflows + +Response: +{ + "workflows": [ + { + "workflow_id": "daily-security-scan", + "schedule": "0 2 * * *", + "is_active": true, + "created": "2025-02-01 15:30:00 UTC" + }, + { + "workflow_id": "hourly-health-check", + "schedule": "0 * * * *", + "is_active": true, + "created": "2025-02-01 16:00:00 UTC" + } + ] +} +``` + +### Cancel Cron Workflow +```bash +DELETE /api/v1/cron/workflows/daily-security-scan + +Response: +{ + "status": "cancelled", + "workflow_id": "daily-security-scan", + "cancelled_at": "2025-02-01 16:30:00 UTC" +} +``` + +--- + +## SUMMARY + +βœ… **llm-router handles both**: +- One-time workflows (instant execution) +- Cron workflows (scheduled, recurring) + +βœ… **Same RoutingWorkflow executor** for both + +βœ… **Temporal manages scheduling** (native support) + +βœ… **Full execution history** tracked + +This completes the architecture! πŸŽ‰ + diff --git a/DESIGN_MASTER_REVIEW.md b/DESIGN_MASTER_REVIEW.md new file mode 100644 index 0000000..2d2ffdb --- /dev/null +++ b/DESIGN_MASTER_REVIEW.md @@ -0,0 +1,888 @@ +# Poimen Service Integration & Dynamic Workflows β€” MASTER REVIEW DOCUMENT + +**Status**: Ready for Approval +**Version**: 1.0 +**Date**: 2025-01-31 +**Duration**: 52-58 hours (4 weeks) + +--- + +## EXECUTIVE SUMMARY (5 MINUTES) + +### Problem +Current workflows are **hardcoded in Go**. Changing activity sequences requires: +- Code modification +- Recompilation +- K8s deployment +- Pod restart +- **Total: 5-10 minutes** + +### Solution +New **RoutingWorkflow** reads JSON WorkflowSpec and executes dynamically: +- **Three entry points**: CLI + HTTP API + Legacy CLI (backward compatible) +- **Three patterns**: Sequential, Await-Task-Complete (KMSvc queue), Retry with error handling +- **JSONPath parameters**: `${step1.output.path}` instead of hardcoded values +- **Result**: Workflow changes in **seconds** (API call only) + +### Impact +βœ… Instant changes | βœ… Three interfaces | βœ… Full compatibility | ❌ +KMSvc complexity + +--- + +## PART 1: THE THREE PATTERNS + +### Pattern 1: Sequential (A β†’ B β†’ C) + +Activities execute in sequence. Each step passes output to next via JSONPath. + +```json +{ + "activities": [ + { + "name": "Clone", + "resource": "CloneRepoActivity", + "parameters": {"repo": "${input.repo}"}, + "timeout": "5m", + "next": "Analyze" + }, + { + "name": "Analyze", + "resource": "AnalyzeCodeActivity", + "parameters": {"path": "${Clone.output.path}"}, + "timeout": "10m", + "next": "Judge" + }, + { + "name": "Judge", + "resource": "JudgeActivity", + "parameters": {"code": "${Analyze.output.code}"}, + "timeout": "5m", + "end": true + } + ] +} +``` + +**Execution Flow**: +``` +Clone ──(output: {path: /tmp/repo})──> Analyze ──(output: {code: ...})──> Judge ──(final result) +``` + +**Use Case**: Code review pipeline (clone β†’ analyze β†’ judge) + +--- + +### Pattern 2: Await-Task-Complete (Launch β†’ Poll Queue β†’ Process) + +Launch long-running job, poll KMSvc queue for result with correlation ID matching, then proceed. + +```json +{ + "name": "LaunchJob", + "type": "awaitTaskComplete", + "resource": "LaunchJobActivity", + "parameters": {"job_id": "job-123"}, + "queue": "job-completions", + "correlation_key": "${LaunchJob.output.correlation_id}", + "timeout": "5m", + "next": "ProcessResult" +} +``` + +**Execution Flow**: +``` +LaunchJobActivity ──(returns {correlation_id: "xyz"})──> Poll Queue "job-completions" + ↓ + [External system completes job] + [Publishes {correlation_id: "xyz", result: ...}] + ↓ + [Workflow receives message] + [Proceeds to ProcessResult] +``` + +**Note**: Pattern 2 type name is `awaitTaskComplete` (awaits task completion via queue) + +**Use Case**: Long-running batch jobs (model training, data processing) + +--- + +### Pattern 3: Retry with Error Handling (Retry N times with backoff) + +Activity retries N times with exponential backoff. On failure, jumps to catch block. + +```json +{ + "name": "Generate", + "resource": "ImplementerActivity", + "parameters": {"task": "${input.task}"}, + "timeout": "30m", + "retry": { + "maxAttempts": 3, + "backoffRate": 2.0, + "initialInterval": "1s" + }, + "catch": [ + {"errorEquals": ["Timeout"], "next": "HandleTimeout"} + ], + "next": "Verify" +} +``` + +**Execution Flow**: +``` +Generate (attempt 1) ──TIMEOUT──> [Wait 1s] +Generate (attempt 2) ──TIMEOUT──> [Wait 2s] +Generate (attempt 3) ──TIMEOUT──> [Wait 4s] +Max retries exceeded ──> Jump to HandleTimeout +``` + +**Use Case**: Code generation (retry on timeout, handle gracefully) + +--- + +## PART 2: THE THREE ENTRY POINTS + +### Entry Point 1: CLI (New Command-Line) + +```bash +# Submit workflow from JSON file +$ poimen-cli submit workflow.json +Workflow submitted: wf-abc123 +Poll status: poimen-cli status wf-abc123 + +# Submit and wait for results +$ poimen-cli submit workflow.json --wait +Waiting for workflow to complete... +βœ… Workflow completed +Output: {...} + +# Execute template immediately (sync, no polling) +$ poimen-cli execute template code-review-v1 --input-file input.json +Executing template... +Step 1/3: Clone ... OK (45s) +Step 2/3: Analyze ... OK (120s) +Step 3/3: Judge ... OK (30s) +βœ… COMPLETED +Output: {...} + +# Check status later +$ poimen-cli status wf-abc123 +Workflow ID: wf-abc123 +Status: COMPLETED +Started: 2025-01-31 15:04:05 +Completed: 2025-01-31 15:05:00 + +# List templates +$ poimen-cli template list +Available Templates: + - code-review-v1 + - clone-analyze-v1 +``` + +**Implementation**: `cmd/cli/` (new directory) + +--- + +### Entry Point 2: HTTP API (New) + +#### Route 1: Submit Workflow (Async) +```bash +curl -X POST http://api.example.com/api/v1/workflows \ + -H "Content-Type: application/json" \ + -d @workflow.json +``` + +**Response**: +```json +{ + "workflow_id": "wf-abc123", + "status": "RUNNING", + "polling_url": "/api/v1/workflows/wf-abc123/status" +} +``` + +#### Check Status +```bash +curl http://api.example.com/api/v1/workflows/wf-abc123/status +``` + +**Response**: +```json +{ + "workflow_id": "wf-abc123", + "status": "COMPLETED", + "steps_completed": 3, + "results": { + "Clone": { "status": "COMPLETED", "output": {...} }, + "Analyze": { "status": "COMPLETED", "output": {...} }, + "Judge": { "status": "COMPLETED", "output": {...} } + }, + "final_output": {...} +} +``` + +#### Route 2: Execute Template (Sync) +```bash +curl -X POST http://api.example.com/api/v1/execute \ + -H "Content-Type: application/json" \ + -d '{ + "template": "code-review-v1", + "input": {"repo": "https://github.com/...", "branch": "feature/x"}, + "timeout": "2m" + }' +``` + +**Response** (immediate): +```json +{ + "status": "COMPLETED", + "results": {...}, + "final_output": {...} +} +``` + +**Implementation**: `cmd/api-server/` (new directory) + +--- + +### Entry Point 3: Legacy CLI (Unchanged) + +```bash +# Old way still works, completely backward compatible +$ go run ./cmd/starter \ + --repo https://github.com/rockliang/poimen \ + --remote file:///tmp/poimen-output \ + --milestone T0 \ + --planner-model ornith \ + --judge-model ornith \ + --implementer-model claude-sonnet-5 + +Workflow submitted: orch-poimen +Status: RUNNING +``` + +**Implementation**: `cmd/starter/` (existing, unchanged) + +--- + +## PART 3: ARCHITECTURE OVERVIEW + +``` +β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” +β”‚ THREE ENTRY POINTS β”‚ +β”œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€ +β”‚ β”‚ +β”‚ CLI (JSON file) HTTP API (JSON POST) Legacy CLI β”‚ +β”‚ β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β” β”‚ +β”‚ β”‚ poimen-cli β”‚ β”‚ /api/v1/ β”‚ β”‚ cmd/ β”‚ β”‚ +β”‚ β”‚ submit β”‚ β”‚ workflows β”‚ β”‚ starter β”‚ β”‚ +β”‚ β””β”€β”€β”€β”€β”€β”€β”¬β”€β”€β”€β”€β”€β”€β”€β”˜ β””β”€β”€β”€β”€β”€β”€β”¬β”€β”€β”€β”€β”€β”€β”€β”˜ β””β”€β”€β”€β”€β”¬β”€β”€β”€β”€β”˜ β”‚ +β”‚ β”‚ β”‚ β”‚ β”‚ +β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”Όβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”Όβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”Όβ”€β”€β”€β”€β”€β”€β”€β”€β”˜ + β”‚ β”‚ β”‚ + β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”Όβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ + ↓ + β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” + β”‚ RoutingWorkflow β”‚ + β”‚ (NEW - generic) β”‚ + β”‚ β”‚ + β”‚ 1. Parse WorkflowSpec β”‚ + β”‚ 2. For each activity: β”‚ + β”‚ - Resolve JSONPath β”‚ + β”‚ - Execute activity β”‚ + β”‚ - Handle errors β”‚ + β”‚ - Store result β”‚ + β”‚ 3. Return results β”‚ + β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”¬β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ + ↓ + β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” + β”‚ Temporal Server β”‚ + β”‚ β”‚ + β”‚ Worker Pool: β”‚ + β”‚ - Executes activities β”‚ + β”‚ - Returns results β”‚ + β”‚ - Retries with backoff β”‚ + β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ +``` + +--- + +## PART 4: BEFORE vs AFTER + +### Before (Hardcoded Workflows) + +```bash +# To change activity sequence: +1. Modify Go code (OrchestratorWorkflow in statemachine/) +2. Compile: go build ./cmd/worker +3. Push docker image +4. Deploy to K8s: kubectl set image deployment/poimen-worker +5. Wait for rollout +Time: 5-10 minutes +``` + +**Workflow definition**: +```go +// Hard to change, tight coupling to specific sequence +if err := workflow.ExecuteActivity(ctx, "CloneRepoActivity", params).Get(ctx, nil); err != nil { + return err +} +if err := workflow.ExecuteActivity(ctx, "PlanningActivity", params2).Get(ctx, nil); err != nil { + return err +} +// ... more activities hardcoded +``` + +--- + +### After (Dynamic Workflows) + +```bash +# To change activity sequence: +1. Edit workflow.json (define sequence) +2. Call API or CLI +3. Done +Time: Seconds +``` + +**Workflow definition**: +```json +{ + "activities": [ + {"name": "Clone", "resource": "CloneRepoActivity", "parameters": {...}, "next": "Analyze"}, + {"name": "Analyze", "resource": "AnalyzeCodeActivity", "parameters": {...}, "next": "Judge"}, + {"name": "Judge", "resource": "JudgeActivity", "parameters": {...}, "end": true} + ] +} +``` + +--- + +## PART 5: IMPLEMENTATION PHASES + +### Phase 1: Routing Engine (Week 1) β€” 20-22 Hours + +**Goal**: Core RoutingWorkflow that executes JSON sequences dynamically + +**Files to Create**: +``` +internal/routing/ + β”œβ”€ types.go (2h) - WorkflowSpec, ActivityStep + β”œβ”€ jsonpath_evaluator.go (3h) - ${step1.output.path} resolution + β”œβ”€ execution_context.go (2h) - State tracking (vars, step results) + β”œβ”€ activity_registry.go (2h) - Activity name β†’ function mapping + β”œβ”€ kmsvc_client.go (2-3h)- KMSvc queue polling (awaitTaskComplete) + └─ *_test.go (tests) + +statemachine/ + β”œβ”€ routing_workflow.go (4-5h)- ⭐ Main workflow logic + └─ routing_workflow_test.go (2h) - Tests + +cmd/worker/main.go: Register RoutingWorkflow (1h) +``` + +**Deliverable**: RoutingWorkflow executes activity sequences with: +- βœ… Sequential execution (A β†’ B β†’ C) +- βœ… Await-Task-Complete pattern (KMSvc queue polling) +- βœ… Retry with exponential backoff +- βœ… Error catch blocks +- βœ… JSONPath parameter resolution + +**Risk**: Low-Medium (KMSvc queue polling adds complexity) + +--- + +### Phase 2: API Server (Week 2) β€” 12-14 Hours + +**Goal**: HTTP endpoints for workflow submission and status polling + +**Files to Create**: +``` +cmd/api-server/ + β”œβ”€ main.go (2h) - Server bootstrap + β”œβ”€ handlers.go (4h) - Routes: /workflows, /execute, status + β”œβ”€ validation.go (2h) - Validate WorkflowSpec + β”œβ”€ template_loader.go (2-3h)- Load predefined templates + └─ *_test.go (2h) - Tests +``` + +**Deliverable**: +- `POST /api/v1/workflows` β€” Submit workflow (async) +- `GET /api/v1/workflows/{id}/status` β€” Check status +- `POST /api/v1/execute` β€” Execute template (sync) + +**Risk**: Low (isolated, no breaking changes) + +--- + +### Phase 3: CLI (Week 2) β€” 8-10 Hours + +**Goal**: Command-line interface for workflow submission + +**Files to Create**: +``` +cmd/cli/ + β”œβ”€ main.go (2h) - Entry point, command routing + └─ commands/ + β”œβ”€ submit.go (2h) - poimen-cli submit workflow.json + β”œβ”€ status.go (2h) - poimen-cli status wf-id + β”œβ”€ template.go (1h) - poimen-cli template list/show + β”œβ”€ execute.go (1h) - poimen-cli execute template + └─ history.go (1h) - poimen-cli history wf-id +``` + +**Deliverable**: +- `poimen-cli submit workflow.json [--wait] [--watch]` +- `poimen-cli status wf-id [--wait]` +- `poimen-cli execute template name --input-file input.json` +- `poimen-cli template list/show` + +**Risk**: Low + +--- + +### Phase 4: Kubernetes Integration (Week 3) β€” 5 Hours + +**Goal**: Deploy api-server to K8s, expose via ingress + +**Files to Create**: +``` +k8s/ + β”œβ”€ api-server-deployment.yaml (2h) - 2 replicas, Temporal connection + β”œβ”€ api-server-service.yaml (1h) - Port 8080 + └─ api-server-networkpolicy.yaml(1h) - Allow ingress traffic + +homelab-frontend/k8s/ + └─ gateway-routes.yaml (1h) - Route /api/v1/workflows to api-server +``` + +**Deliverable**: API exposed at `https://api.riotpiao.com/api/v1/workflows` + +**Risk**: Medium (network policy changes) + +--- + +### Phase 5: Documentation (Week 4) β€” 7 Hours + +**Files to Create**: +``` +docs/ + β”œβ”€ API.md (2h) - Complete API reference + β”œβ”€ CLI.md (2h) - CLI commands + examples + β”œβ”€ TEMPLATES.md (1h) - How to create/manage templates + β”œβ”€ MIGRATION.md (1h) - Migration from cmd/starter + └─ examples/ + β”œβ”€ sequential-workflow.json + β”œβ”€ retry-workflow.json + └─ template-execute.json +``` + +**Deliverable**: Complete documentation with working examples + +**Risk**: Low + +--- + +### Timeline Summary + +``` +Phase 1: Routing Engine 20-22 hours (Week 1) +Phase 2: API Server 12-14 hours (Week 2) +Phase 3: CLI 8-10 hours (Week 2) +Phase 4: K8s Integration 5 hours (Week 3) +Phase 5: Documentation 7 hours (Week 4) +───────────────────────────────────────────────────── +TOTAL 52-58 hours (~1.5 weeks) + +Team: 2-3 engineers +Duration: 4 weeks (with overlapping phases) +``` + +--- + +## PART 6: CRITICAL KMSVC QUESTIONS + +These answers **block implementation** of Phase 1: + +### Q1: What is KMSvc? +- Is it Kafka? Redis? AWS SQS? Custom system? +- What Go client library should we use? +- Example: `github.com/segmentio/kafka-go`? + +### Q2: Message Format +- Is it JSON? +- What fields are required? (e.g., `correlation_id`, `result`, `status`?) +- Example: +```json +{ + "correlation_id": "xyz123", + "status": "completed", + "result": {...} +} +``` + +### Q3: Correlation ID Management +- Who generates the correlation_id? (LaunchJobActivity? Or RoutingWorkflow?) +- Is it returned in LaunchJobActivity output? +- Example: LaunchJobActivity returns: +```json +{ + "correlation_id": "abc123-generated-by-activity", + "job_id": "job-123" +} +``` + +### Q4: Queue Polling Strategy +- Active polling loop (pull messages every N seconds)? OR +- Temporal Signal (push-based, external system sends signal to workflow)? +- If polling: What interval? (1s, 5s, 10s?) + +### Q5: Message Timeout Behavior +- If no message arrives within timeout (e.g., 5m), what happens? + - Option A: Jump to catch block (error handler) + - Option B: Fail entire workflow + - Option C: Infinite wait (only human intervention can stop) + +### Q6: Consumer Group Strategy +- Single shared consumer (all workflows share one connection)? +- Per-workflow consumer (each workflow gets own connection)? +- Dead-letter queue for unmatched messages? + +--- + +## PART 7: KEY DESIGN DECISIONS + +### Decision 1: Three Patterns (Sequential, Await-Task-Complete, Retry) +**Rationale**: Covers 90% of use cases. Retry critical for production. Await-Task-Complete enables long-running jobs. +**Alternative Rejected**: Just Sequential (too limited) + +### Decision 2: KMSvc Queue for Await-Task-Complete Pattern +**Rationale**: Decouples external systems from Temporal. External systems only need queue client, not Temporal SDK. +**Alternative Rejected**: Temporal callback/signal (tight coupling) + +### Decision 3: JSONPath for Parameters +**Rationale**: Standard notation (AWS Step Functions, Kubernetes). Supports nested refs and arrays. +**Alternative Rejected**: Go templating (overkill, security risks) + +### Decision 4: RoutingWorkflow (not Activity) +**Rationale**: Workflows can call activities with retries. Durable execution + replay guarantee. +**Alternative Rejected**: RoutingActivity (can't call other activities, no retries) + +### Decision 5: Static Activity Registry +**Rationale**: Type-safe, simpler. Worker pod must compile all activities anyway. +**Alternative Rejected**: Dynamic registration (complex, unsafe) + +### Decision 6: Keep cmd/starter Forever +**Rationale**: No need to break existing deployments. Let customers choose. +**Alternative Rejected**: Deprecate (breaking change) + +### Decision 7: Shared api-server (2 replicas) +**Rationale**: Simpler, resource-efficient, easier to scale. +**Alternative Rejected**: Per-namespace servers (overkill) + +--- + +## PART 8: RISKS & MITIGATIONS + +| Risk | Severity | Mitigation | +|------|----------|-----------| +| JSONPath evaluation bugs | Medium | Unit tests (>90% coverage), property-based testing | +| **KMSvc queue reliability** | **High** | Dead-letter queue, timeouts, persistence checks | +| **Correlation ID mismatch** | **Medium** | Strict matching, schema validation, detailed logging | +| **Consumer group conflicts** | **Medium** | Clear strategy, rebalancing, message redelivery | +| Activity registry mismatch | Low | Validate at spec submission time | +| Temporal network issues | Low | Retry + timeout in api-server | +| Cache staleness (api-server) | Low | 24h TTL, event-driven invalidation | +| Performance regression | Low | Measure baseline: target <200ms submit, <100ms poll | + +**Total Risk Score**: Medium-High (up from Low-Medium due to KMSvc) + +--- + +## PART 9: SUCCESS CRITERIA + +### Functional βœ… +- RoutingWorkflow executes activity sequences dynamically +- JSONPath parameters resolve correctly (nested, arrays) +- Error catch blocks jump to handlers +- Retry logic with exponential backoff (1s β†’ 2s β†’ 4s) +- API validates WorkflowSpec (no dangling refs) +- CLI reads JSON (no hardcoded flags) +- Status polling returns step-by-step results +- Backward compatible (cmd/starter unchanged) + +### Non-Functional βœ… +- Performance: <200ms workflow submit, <100ms poll +- Uptime: 99.9% (2 replicas, rolling updates) +- Test coverage: >90% +- Documentation: Complete + 3 working examples + +--- + +## PART 10: BACKWARD COMPATIBILITY + +### What Stays Unchanged βœ… +- `cmd/starter/` (old CLI works exactly as before) +- All existing activities +- All existing workflows (OrchestratorWorkflow, TaskUnitWorkflow) +- Temporal worker registration + +### What's New βœ… +- `cmd/cli/` (new CLI) +- `cmd/api-server/` (new HTTP API) +- `internal/routing/` (new routing engine) +- `statemachine/routing_workflow.go` (new workflow) + +### Migration Path βœ… +- Customers can use OLD way or NEW way +- No forced upgrades +- No breaking changes + +--- + +## PART 11: FILES SUMMARY + +### New Files (Phase 1-5) + +**Routing Engine** (Phase 1): +``` +internal/routing/types.go +internal/routing/jsonpath_evaluator.go +internal/routing/execution_context.go +internal/routing/activity_registry.go +internal/routing/kmsvc_client.go +statemachine/routing_workflow.go +``` + +**API Server** (Phase 2): +``` +cmd/api-server/main.go +cmd/api-server/handlers.go +cmd/api-server/validation.go +cmd/api-server/template_loader.go +``` + +**CLI** (Phase 3): +``` +cmd/cli/main.go +cmd/cli/commands/submit.go +cmd/cli/commands/status.go +cmd/cli/commands/template.go +cmd/cli/commands/execute.go +cmd/cli/commands/history.go +``` + +**K8s** (Phase 4): +``` +k8s/api-server-deployment.yaml +k8s/api-server-service.yaml +k8s/api-server-networkpolicy.yaml +homelab-frontend/k8s/gateway-routes.yaml +``` + +**Docs** (Phase 5): +``` +docs/API.md +docs/CLI.md +docs/TEMPLATES.md +docs/MIGRATION.md +docs/examples/*.json +``` + +### Modified Files +- `cmd/worker/main.go` (register RoutingWorkflow) +- `homelab-frontend/k8s/network-policy.yaml` (already done βœ…) + +--- + +## PART 12: COMPARISON TABLE + +| Aspect | Before (Hardcoded) | After (Dynamic) | +|--------|---|---| +| **Workflow Definition** | Go code (OrchestratorWorkflow) | JSON (customer-specified) | +| **Sequencing** | Fixed (Cloneβ†’Planβ†’Implβ†’Judge) | Any (customer defines in JSON) | +| **Parameters** | Direct: `in.RemoteURL` | JSONPath: `${input.repo}` | +| **Change Time** | 5-10 min (recompile+deploy) | Seconds (API call) | +| **Entry Points** | CLI only (hardcoded flags) | CLI + API + Legacy CLI | +| **Error Handling** | Try-catch (per-activity) | Catch blocks (conditional jumps) | +| **Retry Policy** | Temporal default | Configurable per-activity | +| **Long-Running Jobs** | Not supported | awaitTaskComplete pattern | +| **Backward Compat** | N/A | 100% (cmd/starter unchanged) | +| **Code Coupling** | Tight (to specific sequence) | Loose (generic RoutingWorkflow) | + +--- + +## PART 13: APPROVAL CHECKLIST + +### Must Answer (Blocks Phase 1) +- [ ] Q1: What is KMSvc system? +- [ ] Q2: Message format for KMSvc? +- [ ] Q3: Who generates correlation_id? +- [ ] Q4: Polling strategy (active or signal)? +- [ ] Q5: Timeout behavior? +- [ ] Q6: Consumer group strategy? + +### Should Approve +- [ ] Timeline OK? (52-58 hours) +- [ ] Architecture OK? (RoutingWorkflow + 3 entry points) +- [ ] Risk level acceptable? (Medium-High) +- [ ] Backward compatibility OK? (Keep cmd/starter) + +### Sign-Off +- [ ] Tech Lead approval +- [ ] Stakeholder sign-off +- [ ] Ready to start Phase 1 + +--- + +## PART 14: NEXT STEPS + +1. **Review This Document** (1-2 hours) + - Read Parts 1-7 (understanding) + - Read Parts 8-9 (risks & success) + - Read Part 13 (approval checklist) + +2. **Answer Critical Questions** (Q1-Q6) + - Email or Slack responses required + - Needed before Phase 1 can start + +3. **Stakeholder Approval** + - Architecture review meeting + - Risk acknowledgment + - Sign-off on timeline + +4. **Start Implementation** + - Week 1: Phase 1 (RoutingWorkflow) + - Week 2: Phase 2 (API) + Phase 3 (CLI) + - Week 3: Phase 4 (K8s deploy) + - Week 4: Phase 5 (Docs) + +--- + +## APPENDIX: EXAMPLE WORKFLOW JSON + +```json +{ + "name": "code-review-pipeline", + "input": { + "repo": "https://github.com/rockliang/poimen", + "branch": "feature/new-cli" + }, + "activities": [ + { + "name": "Clone", + "resource": "CloneRepoActivity", + "parameters": { + "repo": "${input.repo}", + "branch": "${input.branch}" + }, + "timeout": "5m", + "next": "Analyze" + }, + { + "name": "Analyze", + "resource": "AnalyzeCodeActivity", + "parameters": { + "path": "${Clone.output.path}" + }, + "timeout": "10m", + "retry": { + "maxAttempts": 3, + "backoffRate": 2.0, + "initialInterval": "1s" + }, + "catch": [ + { + "errorEquals": ["Timeout"], + "next": "HandleTimeout" + } + ], + "next": "Judge" + }, + { + "name": "Judge", + "resource": "JudgeActivity", + "parameters": { + "code": "${Analyze.output.code}", + "analysis": "${Analyze.output.analysis}" + }, + "timeout": "5m", + "end": true + }, + { + "name": "HandleTimeout", + "type": "pass", + "result": { + "status": "failed", + "reason": "Analysis timed out after 10m" + }, + "end": true + } + ] +} +``` + +--- + +## APPENDIX: CLI USAGE EXAMPLES + +```bash +# Submit workflow from JSON file +$ poimen-cli submit workflow.json +Workflow submitted: wf-abc123 +Poll status: poimen-cli status wf-abc123 + +# Submit and wait for results +$ poimen-cli submit workflow.json --wait +Waiting for workflow to complete... +βœ… Workflow completed +Output: +{ + "verdict": "approved", + "score": 0.95, + "issues": [] +} + +# Check status +$ poimen-cli status wf-abc123 +Workflow ID: wf-abc123 +Status: COMPLETED +Started: 2025-01-31 15:04:05 +Completed: 2025-01-31 15:05:00 + +# Execute template immediately (sync) +$ poimen-cli execute template code-review-v1 --input-file input.json +Executing template: code-review-v1 +[15:04:05] Step 1/3: Clone ... OK (45s) +[15:04:50] Step 2/3: Analyze ... OK (120s) +[15:05:50] Step 3/3: Judge ... OK (30s) +βœ… COMPLETED +Output: {...} + +# List templates +$ poimen-cli template list +Available Templates: + - code-review-v1 + - clone-analyze-v1 + - simple-test-v1 + +# Show template +$ poimen-cli template show code-review-v1 +Template: code-review-v1 +Activities: 3 + +1. Clone (CloneRepoActivity) + Timeout: 5m +2. Analyze (AnalyzeCodeActivity) + Timeout: 10m + Retry: 3 attempts, backoff 2.0 +3. Judge (JudgeActivity) + Timeout: 5m +``` + +--- + +**END OF MASTER REVIEW DOCUMENT** + +For detailed technical documentation, see existing design docs in the repo. + diff --git a/IMPLEMENTATION_TASKS.md b/IMPLEMENTATION_TASKS.md new file mode 100644 index 0000000..40a675a --- /dev/null +++ b/IMPLEMENTATION_TASKS.md @@ -0,0 +1,1158 @@ +# Poimen Routing Workflow - Implementation Tasks + +**Total Effort**: 60-70 hours (2-3 weeks, 1-2 engineers) + +--- + +## PHASE 1: FOUNDATION (8-10 hours) + +### Task 1.1: Create Go Type Definitions + +**File**: `internal/routing/types.go` + +**What to create**: +```go +// Core types +type WorkflowSpec struct { + Name string + Input map[string]interface{} + States []State +} + +type State struct { + Name string + Type StateType + Resource string + Parameters map[string]interface{} + Timeout string + Retry *RetryPolicy + Catch []CatchClause + Result interface{} + Error, Cause string + Next string + End bool +} + +type StateType string +const ( + StateTypeTask StateType = "Task" + StateTypePass StateType = "Pass" + StateTypeFail StateType = "Fail" +) + +type RetryPolicy struct { + MaxAttempts int32 + BackoffRate float64 + InitialInterval string + MaxInterval string +} + +type CatchClause struct { + ErrorEquals []string + Next string +} + +// Cron types +type CronWorkflowSpec struct { + Name string + Type string // "CronWorkflow" + Schedule string // "0 2 * * *" + Timezone string + Input map[string]interface{} + States []State + MaxConcurrent int + Timeout string + EnableHistory bool +} + +type ExecutionContext struct { + Input map[string]interface{} + StepResults map[string]interface{} +} + +type Result struct { + FinalOutput interface{} + Status string + Error error +} +``` + +**Acceptance Criteria**: +- [ ] All types compile +- [ ] JSON marshaling/unmarshaling works +- [ ] Unit tests for type serialization + +**Effort**: 2 hours + +--- + +### Task 1.2: Create ActivityKnowledgeBase.json + +**File**: `internal/routing/activity_knowledge_base.json` + +**What to create**: +```json +{ + "activities": [ + { + "name": "CloneRepoActivity", + "description": "Clone a Git repository", + "category": "source-control", + "inputs": { + "repo": {"type": "string", "required": true}, + "branch": {"type": "string", "required": false, "default": "main"} + }, + "outputs": { + "path": {"type": "string"}, + "commit": {"type": "string"} + }, + "constraints": { + "defaultTimeout": "5m", + "isFlaky": false, + "recommendedRetries": 2, + "retryBackoff": 1.5 + } + }, + { + "name": "AnalyzeCodeActivity", + "description": "Analyze code for quality", + "category": "analysis", + "inputs": { + "path": {"type": "string", "required": true} + }, + "outputs": { + "quality": {"type": "number"}, + "issues": {"type": "array"} + }, + "constraints": { + "defaultTimeout": "10m", + "isFlaky": true, + "recommendedRetries": 3, + "retryBackoff": 2.0 + } + }, + // Add 5-7 more activities (SecurityScanActivity, PerformanceAnalysisActivity, etc) + ] +} +``` + +**Acceptance Criteria**: +- [ ] Valid JSON format +- [ ] At least 7 activities defined +- [ ] Each has description, inputs, outputs, constraints +- [ ] Includes both flaky and non-flaky activities +- [ ] JSON loads without error + +**Effort**: 3 hours + +--- + +### Task 1.3: Create ActivityKnowledgeBase Loader + +**File**: `internal/routing/activity_knowledge_base.go` + +**What to create**: +```go +type ActivityKnowledgeBase struct { + Activities map[string]*ActivityMetadata +} + +type ActivityMetadata struct { + Name string + Description string + Category string + Inputs map[string]InputField + Outputs map[string]OutputField + Constraints Constraints +} + +type Constraints struct { + DefaultTimeout string + IsFlaky bool + RecommendedRetries int + RetryBackoff float64 +} + +// Load from JSON file +func LoadActivityKnowledgeBase(path string) (*ActivityKnowledgeBase, error) { + // Load JSON from file + // Parse into ActivityKnowledgeBase + // Return +} + +// Get activity by name +func (kb *ActivityKnowledgeBase) GetActivity(name string) *ActivityMetadata + +// Get all activities in category +func (kb *ActivityKnowledgeBase) GetByCategory(category string) []*ActivityMetadata +``` + +**Acceptance Criteria**: +- [ ] Loads JSON successfully +- [ ] GetActivity() works +- [ ] GetByCategory() works +- [ ] Unit tests pass + +**Effort**: 2 hours + +--- + +### Task 1.4: Create WorkflowSpec Validator + +**File**: `internal/routing/validator.go` + +**What to validate**: +```go +func ValidateWorkflowSpec(spec *WorkflowSpec) error { + // 1. All states have unique names + // 2. All state.next references exist + // 3. All catch.next references exist + // 4. All Task states have resource defined + // 5. No circular loops (Aβ†’Bβ†’Cβ†’A) + // 6. At least one path leads to end + // 7. No dead ends + // 8. Timeout format valid (e.g., "5m") + // 9. Retry settings valid + // 10. JSONPath expressions syntactically valid +} + +func ValidateCronExpression(schedule string) error { + // Use cronexpr library + // Parse and validate +} +``` + +**Acceptance Criteria**: +- [ ] Catches all 10 validation errors +- [ ] Unit tests for each error type +- [ ] Cron expression validation works +- [ ] Clear error messages + +**Effort**: 3 hours + +--- + +## PHASE 2: LLM-ROUTER ACTIVITY (12-15 hours) + +### Task 2.1: Create JSONPath Resolver + +**File**: `internal/routing/jsonpath_evaluator.go` + +**What to create**: +```go +func ResolveJSONPath(expr interface{}, ec *ExecutionContext) interface{} { + // Support: + // ${input.repo} β†’ from ec.Input + // ${Clone.output.path} β†’ from ec.StepResults["Clone"]["output"]["path"] + // ${input.branch:main} β†’ default value "main" +} + +func ResolveParameters(params map[string]interface{}, ec *ExecutionContext) map[string]interface{} { + // Resolve all JSONPath expressions in params +} + +func getNestedValue(obj interface{}, path []string) interface{} { + // Handle nested object access +} +``` + +**Test cases**: +```go +// Test: ${input.repo} +// Test: ${Clone.output.path} +// Test: ${Analyze.output.issues[0]} +// Test: ${input.branch:main} with default +// Test: Literal values (no JSONPath) +// Test: Nested objects +``` + +**Acceptance Criteria**: +- [ ] All JSONPath expressions resolve correctly +- [ ] Default values work +- [ ] Array access works +- [ ] Nested objects work +- [ ] 15+ unit tests pass + +**Effort**: 3 hours + +--- + +### Task 2.2: Create llm-router Activity Skeleton + +**File**: `cmd/worker/activities/llm_router.go` + +**What to create**: +```go +type LLMRouterInput struct { + Message string + Repo string + Branch string +} + +func LLMRouterActivity(ctx context.Context, input LLMRouterInput) (routing.WorkflowSpec, error) { + // 1. Load activity knowledge base + kb, err := routing.LoadActivityKnowledgeBase() + + // 2. Create LLM prompt + // (use memory service to call LLM) + + // 3. Parse LLM response to understand intent + intent := parseLLMResponse(llmOutput) + + // 4. Build workflow spec + spec := buildWorkflowSpec(intent, input, kb) + + // 5. Validate spec + if err := routing.ValidateWorkflowSpec(&spec); err != nil { + return routing.WorkflowSpec{}, err + } + + return spec, nil +} +``` + +**Acceptance Criteria**: +- [ ] Activity compiles +- [ ] Loads knowledge base successfully +- [ ] Can be registered in worker + +**Effort**: 2 hours + +--- + +### Task 2.3: Implement LLM Intent Analysis + +**File**: `cmd/worker/activities/llm_intent.go` + +**What to analyze**: +```go +type Intent struct { + IsScheduled bool + Schedule string // Cron expression if scheduled + Timezone string + SelectedActivities []string + ActivityDependencies map[string][]string + NeedsFinalJudgment bool + NeedsNotification bool +} + +func analyzeLLMIntent(ctx context.Context, kb *ActivityKnowledgeBase, message string) (Intent, error) { + // Use LLM to understand: + // 1. Is this a scheduled job? Extract cron expression + // 2. Which activities are needed? + // 3. What order? + // 4. Does it need error handling? + // 5. What parameters to pass? +} + +func extractCronFromMessage(message string) string { + // Use LLM to convert: + // "every day at 2 AM" β†’ "0 2 * * *" + // "every 6 hours" β†’ "0 */6 * * *" + // etc +} +``` + +**Acceptance Criteria**: +- [ ] Detects scheduled workflows (contains "daily", "every", etc) +- [ ] Extracts cron expressions correctly +- [ ] Identifies required activities +- [ ] Orders activities by dependencies +- [ ] Integration tests with real LLM calls + +**Effort**: 5 hours + +--- + +### Task 2.4: Implement Workflow Spec Builder + +**File**: `cmd/worker/activities/spec_builder.go` + +**What to build**: +```go +func buildWorkflowSpec(intent Intent, input LLMRouterInput, kb *ActivityKnowledgeBase) routing.WorkflowSpec { + // 1. Create initial states (always Clone first) + // 2. Add selected activities in order + // 3. For each activity: + // - Get timeout from knowledge base + // - Get retry settings from knowledge base + // - If flaky: add error handling + // - Chain parameters (JSONPath) + // 4. Add final state (notification or pass) + // 5. Add error handlers + // 6. Return complete spec +} + +func createStateForActivity(activity *ActivityMetadata, previousState string, nextStateName string) routing.State { + // Create a Task state with intelligent settings +} + +func chainParameters(activity *ActivityMetadata, previousStates map[string]interface{}) map[string]interface{} { + // Use LLM to map outputs from previous steps to inputs of this activity + // E.g.: Clone.output.path β†’ AnalyzeCodeActivity.path +} +``` + +**Acceptance Criteria**: +- [ ] Generates valid WorkflowSpec +- [ ] States ordered correctly +- [ ] Timeout from knowledge base +- [ ] Retry policy from knowledge base +- [ ] Error handlers for flaky activities +- [ ] JSONPath chaining works +- [ ] 10+ integration tests + +**Effort**: 4 hours + +--- + +### Task 2.5: Implement Cron Workflow Generation + +**File**: `cmd/worker/activities/cron_builder.go` + +**What to build**: +```go +func buildCronWorkflowSpec(intent Intent, input LLMRouterInput, kb *ActivityKnowledgeBase) routing.CronWorkflowSpec { + // 1. Extract schedule and timezone + // 2. Build states (same as regular workflow) + // 3. Create CronWorkflowSpec + // 4. Return +} + +func validateCronSchedule(schedule string) error { + // Use cronexpr library + // Validate cron expression +} +``` + +**Acceptance Criteria**: +- [ ] Generates valid CronWorkflowSpec +- [ ] Schedule is valid cron expression +- [ ] Timezone is valid +- [ ] Same states as one-time workflow +- [ ] 5+ test cases + +**Effort**: 2 hours + +--- + +## PHASE 3: ROUTING WORKFLOW EXECUTOR (15-18 hours) + +### Task 3.1: Create State Executor Dispatcher + +**File**: `statemachine/routing_workflow.go` (Part 1) + +**What to create**: +```go +func ExecuteState(ctx workflow.Context, state *routing.State, ec *routing.ExecutionContext) (interface{}, error) { + switch state.Type { + case routing.StateTypeTask: + return ExecuteTaskState(ctx, state, ec) + case routing.StateTypePass: + return ExecutePassState(ctx, state, ec) + case routing.StateTypeFail: + return ExecuteFailState(ctx, state, ec) + default: + return nil, fmt.Errorf("unknown state type: %s", state.Type) + } +} +``` + +**Acceptance Criteria**: +- [ ] Compiles +- [ ] Routes to correct executor +- [ ] Unit tests for each type + +**Effort**: 1 hour + +--- + +### Task 3.2: Implement Task Executor + +**File**: `internal/routing/task_executor.go` + +**What to create**: +```go +func ExecuteTaskState(ctx workflow.Context, state *routing.State, ec *routing.ExecutionContext) (interface{}, error) { + // 1. Resolve JSONPath in parameters + resolvedParams := routing.ResolveJSONPath(state.Parameters, ec) + + // 2. Parse timeout + timeout := routing.ParseDuration(state.Timeout) + + // 3. Build activity options + options := workflow.ActivityOptions{ + StartToCloseTimeout: timeout, + } + + // 4. Add retry policy if specified + if state.Retry != nil { + options.RetryPolicy = &temporal.RetryPolicy{ + InitialInterval: routing.ParseDuration(state.Retry.InitialInterval), + BackoffCoefficient: state.Retry.BackoffRate, + MaximumInterval: routing.ParseDuration(state.Retry.MaxInterval), + MaximumAttempts: state.Retry.MaxAttempts, + } + } + + // 5. Execute activity + actCtx := workflow.WithActivityOptions(ctx, options) + var result interface{} + err := workflow.ExecuteActivity( + actCtx, + state.Resource, + resolvedParams, + ).Get(actCtx, &result) + + return result, err +} +``` + +**Acceptance Criteria**: +- [ ] Resolves JSONPath correctly +- [ ] Parses timeout/retry +- [ ] Executes activity +- [ ] Returns result or error +- [ ] Unit tests + +**Effort**: 2 hours + +--- + +### Task 3.3: Implement Pass & Fail Executors + +**File**: `internal/routing/pass_fail_executor.go` + +**What to create**: +```go +func ExecutePassState(ctx workflow.Context, state *routing.State, ec *routing.ExecutionContext) (interface{}, error) { + return state.Result, nil +} + +func ExecuteFailState(ctx workflow.Context, state *routing.State, ec *routing.ExecutionContext) (interface{}, error) { + return nil, fmt.Errorf("%s: %s", state.Error, state.Cause) +} +``` + +**Acceptance Criteria**: +- [ ] Pass returns static result +- [ ] Fail returns error +- [ ] Unit tests + +**Effort**: 1 hour + +--- + +### Task 3.4: Implement Main RoutingWorkflow + +**File**: `statemachine/routing_workflow.go` (Part 2) + +**What to create**: +```go +func RoutingWorkflow(ctx workflow.Context, spec routing.WorkflowSpec) (routing.Result, error) { + logger := workflow.GetLogger(ctx) + + ec := &routing.ExecutionContext{ + Input: spec.Input, + StepResults: make(map[string]interface{}), + } + + currentStateName := routing.FindFirstState(spec.States) + + for { + logger.Info("Executing state", "state", currentStateName) + + state := routing.FindStateByName(spec.States, currentStateName) + if state == nil { + return routing.Result{}, fmt.Errorf("state not found: %s", currentStateName) + } + + result, err := routing.ExecuteState(ctx, state, ec) + + if err != nil { + // Check catch blocks + handled := false + for _, catchClause := range state.Catch { + if routing.MatchesError(err, catchClause.ErrorEquals) { + logger.Info("Error caught", "handler", catchClause.Next) + currentStateName = catchClause.Next + handled = true + break + } + } + + if !handled { + return routing.Result{ + FinalOutput: ec.StepResults, + Status: "FAILED", + Error: err, + }, nil + } + } else { + ec.StepResults[state.Name] = result + + if state.End { + logger.Info("Workflow completed") + return routing.Result{ + FinalOutput: result, + Status: "COMPLETED", + Error: nil, + }, nil + } + + if state.Next == "" { + return routing.Result{}, fmt.Errorf("state %s has no next and end != true", currentStateName) + } + + currentStateName = state.Next + } + } +} +``` + +**Acceptance Criteria**: +- [ ] Compiles +- [ ] State machine loop works +- [ ] Error handling with catch blocks +- [ ] Transitions correct +- [ ] 20+ unit tests + +**Effort**: 4 hours + +--- + +### Task 3.5: Register RoutingWorkflow in Worker + +**File**: `cmd/worker/main.go` + +**What to do**: +```go +func main() { + // Existing worker setup + + // Register RoutingWorkflow + w.RegisterWorkflow(statemachine.RoutingWorkflow) + + // Register all activities (CloneRepoActivity, AnalyzeCodeActivity, etc) + w.RegisterActivity(CloneRepoActivity) + w.RegisterActivity(AnalyzeCodeActivity) + w.RegisterActivity(SecurityScanActivity) + // ... more activities + + // Register llm-router activity + w.RegisterActivity(activities.LLMRouterActivity) + + // Run worker + err := w.Run() +} +``` + +**Acceptance Criteria**: +- [ ] Worker starts without error +- [ ] All workflows/activities registered +- [ ] Can connect to Temporal +- [ ] Health check passes + +**Effort**: 1 hour + +--- + +### Task 3.6: Implement Helper Functions + +**File**: `internal/routing/helpers.go` + +**What to create**: +```go +func FindFirstState(states []State) string { + // Find state with no incoming edges +} + +func FindStateByName(states []State, name string) *State { + // Find state by name +} + +func ParseDuration(d string) time.Duration { + // Parse "5m" β†’ 5 minutes + // Parse "1h" β†’ 1 hour +} + +func MatchesError(err error, errorTypes []string) bool { + // Check if error matches any of the error types +} +``` + +**Acceptance Criteria**: +- [ ] All helpers work correctly +- [ ] Unit tests + +**Effort**: 2 hours + +--- + +## PHASE 4: API SERVER & CLI (12-15 hours) + +### Task 4.1: Create API Handlers + +**File**: `cmd/api-server/handlers.go` + +**What to create**: +```go +// POST /api/v1/workflows (submit one-time workflow) +func SubmitWorkflowHandler(w http.ResponseWriter, r *http.Request) + +// GET /api/v1/workflows/{id}/status (check status) +func GetWorkflowStatusHandler(w http.ResponseWriter, r *http.Request) + +// POST /api/v1/cron/workflows (submit cron workflow) +func SubmitCronWorkflowHandler(w http.ResponseWriter, r *http.Request) + +// GET /api/v1/cron/workflows/{id}/status (check cron status) +func GetCronStatusHandler(w http.ResponseWriter, r *http.Request) + +// DELETE /api/v1/cron/workflows/{id} (cancel cron) +func CancelCronWorkflowHandler(w http.ResponseWriter, r *http.Request) +``` + +**Acceptance Criteria**: +- [ ] All endpoints compile +- [ ] Input validation +- [ ] Error handling +- [ ] Response formatting +- [ ] 10+ integration tests + +**Effort**: 4 hours + +--- + +### Task 4.2: Create CLI Commands + +**File**: `cmd/cli/main.go` and `cmd/cli/commands/` + +**What to create**: +```bash +# Submit one-time workflow +poimen-cli submit workflow.json --wait + +# Check status +poimen-cli status wf-abc123 --wait + +# Submit cron workflow +poimen-cli cron submit schedule.json + +# List cron workflows +poimen-cli cron list + +# Cancel cron +poimen-cli cron cancel daily-security-scan + +# Execute template (sync) +poimen-cli execute template code-review-v1 --input input.json +``` + +**Files needed**: +``` +cmd/cli/main.go +cmd/cli/commands/submit.go +cmd/cli/commands/status.go +cmd/cli/commands/cron.go +cmd/cli/commands/template.go +``` + +**Acceptance Criteria**: +- [ ] All commands compile +- [ ] Help text works +- [ ] Argument parsing +- [ ] JSON output +- [ ] 10+ e2e tests + +**Effort**: 5 hours + +--- + +### Task 4.3: API Server Bootstrap + +**File**: `cmd/api-server/main.go` + +**What to create**: +```go +func main() { + // Load config + // Create Temporal client + // Create HTTP server + // Register handlers + // Start server +} +``` + +**Acceptance Criteria**: +- [ ] Server starts +- [ ] Listens on correct port +- [ ] Can connect to Temporal +- [ ] Health check endpoint works + +**Effort**: 2 hours + +--- + +### Task 4.4: Input Validation + +**File**: `cmd/api-server/validation.go` + +**What to validate**: +```go +func ValidateWorkflowInput(req WorkflowRequest) error { + // Check: Is JSON valid? + // Check: Does it conform to WorkflowSpec? + // Check: Are all activities registered? + // Check: Is it safe to execute? +} +``` + +**Acceptance Criteria**: +- [ ] Rejects invalid JSON +- [ ] Rejects missing required fields +- [ ] Rejects unknown activities +- [ ] Clear error messages + +**Effort**: 2 hours + +--- + +## PHASE 5: TESTING & INTEGRATION (8-12 hours) + +### Task 5.1: Unit Tests for Types & Validators + +**Files**: +- `internal/routing/types_test.go` +- `internal/routing/validator_test.go` +- `internal/routing/jsonpath_evaluator_test.go` + +**What to test**: +``` +βœ… Type marshaling/unmarshaling +βœ… Validator catches all errors +βœ… JSONPath resolution (15+ cases) +βœ… Cron expression validation +βœ… Activity knowledge base loading +``` + +**Acceptance Criteria**: +- [ ] >90% code coverage +- [ ] All edge cases covered +- [ ] CI passes + +**Effort**: 3 hours + +--- + +### Task 5.2: Integration Tests for Workflows + +**Files**: +- `statemachine/routing_workflow_test.go` +- `cmd/worker/activities/llm_router_test.go` + +**What to test**: +``` +βœ… Simple sequential workflow (Clone β†’ Analyze β†’ Judge) +βœ… Workflow with retry (timeout on attempt 1, success on attempt 2) +βœ… Workflow with error handling (error caught, jump to handler) +βœ… Workflow with pass state +βœ… LLM-router generates correct spec +βœ… LLM-router detects cron jobs +``` + +**Acceptance Criteria**: +- [ ] All scenarios pass +- [ ] Tests run in <30 seconds +- [ ] No flaky tests + +**Effort**: 4 hours + +--- + +### Task 5.3: End-to-End Tests + +**Files**: +- `tests/e2e/workflow_test.go` +- `tests/e2e/api_test.go` +- `tests/e2e/cli_test.go` + +**What to test**: +``` +βœ… User β†’ API β†’ llm-router β†’ RoutingWorkflow β†’ Result +βœ… User β†’ CLI β†’ llm-router β†’ RoutingWorkflow β†’ Result +βœ… Cron workflow submission and scheduling +βœ… Status polling +βœ… Error scenarios (network, timeout, activity failure) +``` + +**Acceptance Criteria**: +- [ ] All flows work end-to-end +- [ ] Can interact with real Temporal +- [ ] Tests run in <2 minutes + +**Effort**: 4 hours + +--- + +### Task 5.4: Load & Performance Tests + +**Files**: +- `tests/load/workflow_load_test.go` + +**What to test**: +``` +βœ… Submit 100 workflows concurrently +βœ… Poll status for 1000 workflows +βœ… Execute 10 sequential workflows +βœ… Measure latency: submit β†’ first activity +βœ… Measure throughput: workflows/second +``` + +**Acceptance Criteria**: +- [ ] Submit latency <200ms +- [ ] Poll latency <100ms +- [ ] Throughput >10 workflows/sec +- [ ] No memory leaks + +**Effort**: 2 hours + +--- + +## PHASE 6: DOCUMENTATION & DEPLOYMENT (5-8 hours) + +### Task 6.1: API Documentation + +**File**: `docs/API.md` + +**What to document**: +``` +- All endpoints +- Request/response formats +- Error codes +- Examples (curl, Python, Go) +- Authentication +- Rate limits +``` + +**Acceptance Criteria**: +- [ ] Complete +- [ ] All examples work +- [ ] Clear and concise + +**Effort**: 2 hours + +--- + +### Task 6.2: CLI Documentation + +**File**: `docs/CLI.md` + +**What to document**: +``` +- All commands +- Usage examples +- Flags and options +- Output formats +- Troubleshooting +``` + +**Acceptance Criteria**: +- [ ] Complete +- [ ] All commands documented +- [ ] Examples work + +**Effort**: 1 hour + +--- + +### Task 6.3: Deployment Guide + +**File**: `docs/DEPLOYMENT.md` + +**What to document**: +``` +- Build instructions +- Docker image +- K8s deployment (api-server, worker) +- Configuration +- Troubleshooting +- Monitoring +``` + +**Acceptance Criteria**: +- [ ] Complete +- [ ] Can deploy from instructions +- [ ] Includes health checks + +**Effort**: 2 hours + +--- + +### Task 6.4: User Guide & Examples + +**File**: `docs/USER_GUIDE.md` + +**What to include**: +``` +- Getting started +- 5+ workflow examples (code review, security scan, health check, etc) +- How to create custom activities +- How to use cron jobs +- FAQ +``` + +**Acceptance Criteria**: +- [ ] Complete +- [ ] All examples runnable +- [ ] Clear explanations + +**Effort**: 2 hours + +--- + +## TIMELINE & PHASES + +``` +Phase 1: Foundation (8-10 hours) + β”œβ”€ Types, Knowledge Base, Validator + └─ Week 1 (Mon-Tue) + +Phase 2: LLM-Router (12-15 hours) + β”œβ”€ Intent Analysis, Spec Builder, Cron Builder + └─ Week 1 (Wed-Fri) + Week 2 (Mon) + +Phase 3: RoutingWorkflow (15-18 hours) + β”œβ”€ Executors, State Machine, Registration + └─ Week 2 (Tue-Thu) + +Phase 4: API/CLI (12-15 hours) + β”œβ”€ Handlers, Commands, Validation + └─ Week 2 (Fri) + Week 3 (Mon-Tue) + +Phase 5: Testing (8-12 hours) + β”œβ”€ Unit, Integration, E2E, Load tests + └─ Week 3 (Wed-Fri) + +Phase 6: Documentation (5-8 hours) + β”œβ”€ API, CLI, Deployment, User Guide + └─ Week 3 (Fri) + Week 4 (Mon) + +TOTAL: 60-70 hours (3 weeks, 1-2 engineers) +``` + +--- + +## DEPENDENCIES + +``` +Task 1.1 Types β†’ Required by all others +Task 1.2 Knowledge Base β†’ Required by Task 2.x +Task 1.3 KB Loader β†’ Required by Task 2.x +Task 1.4 Validator β†’ Required by Task 2.x & 3.x + ↓ +Task 2.x LLM-Router β†’ Required by Task 3.x + ↓ +Task 3.x RoutingWorkflow β†’ Required by Task 4.x & 5.x + ↓ +Task 4.x API/CLI β†’ Can run in parallel with Task 5.x + ↓ +Task 5.x Testing β†’ Gate to Task 6.x + ↓ +Task 6.x Documentation β†’ Final deliverable +``` + +--- + +## RESOURCE ALLOCATION (2-3 Engineers) + +### Engineer 1: Core Workflows +- Tasks: 1.1-1.4 (Foundation) +- Tasks: 2.1-2.5 (LLM-Router) +- Tasks: 3.1-3.6 (RoutingWorkflow) + +### Engineer 2: API/CLI/Testing +- Tasks: 4.1-4.4 (API/CLI) +- Tasks: 5.1-5.4 (Testing) +- Tasks: 6.1-6.4 (Documentation) + +### Both Engineers +- Daily sync on blockers +- Pair programming for complex tasks (2.3, 3.4) +- Code reviews + +--- + +## SUCCESS CRITERIA (Acceptance for Each Phase) + +**Phase 1**: All types compile, KB loads, validator catches errors βœ… +**Phase 2**: llm-router generates valid specs, detects cron βœ… +**Phase 3**: RoutingWorkflow executes any spec, handles errors βœ… +**Phase 4**: API/CLI functional, all endpoints work βœ… +**Phase 5**: >90% test coverage, all scenarios pass βœ… +**Phase 6**: Complete documentation, can deploy βœ… + +--- + +## BLOCKERS TO WATCH + +1. **LLM Integration** (Task 2.3) + - Need: Memory service API working + - Risk: LLM quality + - Mitigation: Test with real LLM early + +2. **Temporal Cron Support** (Task 3.4) + - Need: CronSchedule in StartWorkflowOptions + - Risk: Version compatibility + - Mitigation: Verify Temporal SDK version + +3. **Activity Registration** (Task 3.5) + - Need: All activities implemented + - Risk: Missing activities + - Mitigation: Create stub activities first + +4. **Performance** (Task 5.4) + - Need: <200ms submit latency + - Risk: RoutingWorkflow overhead + - Mitigation: Profile early and often + +--- + +## DELIVERABLES + +### Code Deliverables +- βœ… internal/routing/ (types, validators, executors) +- βœ… statemachine/routing_workflow.go +- βœ… cmd/worker/activities/llm_router.go +- βœ… cmd/api-server/ (full HTTP server) +- βœ… cmd/cli/ (full CLI) +- βœ… tests/ (unit, integration, e2e, load tests) + +### Documentation Deliverables +- βœ… docs/API.md +- βœ… docs/CLI.md +- βœ… docs/DEPLOYMENT.md +- βœ… docs/USER_GUIDE.md +- βœ… docs/examples/ (10+ workflow examples) + +### Operational Deliverables +- βœ… internal/routing/activity_knowledge_base.json +- βœ… k8s/api-server-deployment.yaml +- βœ… Docker image for api-server + +--- + +## DONE CRITERIA + +- [ ] All code compiles without warnings +- [ ] All tests pass (unit, integration, e2e, load) +- [ ] >90% code coverage +- [ ] Documentation complete +- [ ] Can deploy to Kubernetes +- [ ] Can submit workflow from API/CLI +- [ ] Can create cron workflow +- [ ] Performance targets met +- [ ] No critical bugs + +This is **ready to implement**! πŸš€ + diff --git a/README_IMPLEMENTATION.md b/README_IMPLEMENTATION.md new file mode 100644 index 0000000..324d326 --- /dev/null +++ b/README_IMPLEMENTATION.md @@ -0,0 +1,360 @@ +# Poimen Routing Workflow - Complete Implementation Guide + +## πŸ“š DOCUMENTATION STRUCTURE + +You now have **4 complete documents** that form a complete specification: + +### 1. `ROUTING_WORKFLOW_SPEC.md` (30+ KB) +**The Technical Specification** - Everything about the system design + +Contains: +- ActivityKnowledgeBase.json format +- llm-router Activity (intelligent workflow generator) +- RoutingWorkflow (generic executor) +- Go type definitions (copy-paste ready) +- Implementation architecture +- State types (Task, Pass, Fail) +- CronWorkflowSpec (scheduled workflows) +- Execution flow examples +- Cron syntax reference + +**When to use**: Building the system, understanding architecture + +--- + +### 2. `CRON_JOBS_QUICK_REFERENCE.md` (4 KB) +**Quick Reference for Cron Jobs** + +Contains: +- Cron syntax examples +- How llm-router detects scheduled jobs +- Execution tracking +- API endpoints for cron +- One-time vs Cron comparison + +**When to use**: Testing cron features, quick lookup + +--- + +### 3. `IMPLEMENTATION_TASKS.md` (30+ KB) +**The Complete Task Breakdown** - What to build, in what order + +Contains: +- 27 specific, actionable tasks +- Effort estimates per task (2-5 hours each) +- Acceptance criteria for each task +- Dependencies between tasks +- Timeline (3 weeks, 1-2 engineers) +- Resource allocation +- Blockers to watch +- Success criteria + +**Structure**: +``` +Phase 1: Foundation (8-10 hours) + β”œβ”€ Task 1.1: Types + β”œβ”€ Task 1.2: Knowledge Base + β”œβ”€ Task 1.3: KB Loader + └─ Task 1.4: Validator + +Phase 2: LLM-Router (12-15 hours) + β”œβ”€ Task 2.1: JSONPath Resolver + β”œβ”€ Task 2.2: Activity Skeleton + β”œβ”€ Task 2.3: Intent Analysis + β”œβ”€ Task 2.4: Spec Builder + └─ Task 2.5: Cron Builder + +Phase 3: RoutingWorkflow (15-18 hours) + β”œβ”€ Task 3.1-3.6: Executors & State Machine + +Phase 4: API/CLI (12-15 hours) + β”œβ”€ Task 4.1-4.4: Handlers, Commands, Validation + +Phase 5: Testing (8-12 hours) + β”œβ”€ Task 5.1-5.4: Unit, Integration, E2E, Load tests + +Phase 6: Docs & Deployment (5-8 hours) + β”œβ”€ Task 6.1-6.4: API.md, CLI.md, Deployment.md, User Guide +``` + +**When to use**: Planning sprints, assigning work, tracking progress + +--- + +### 4. `DESIGN_MASTER_REVIEW.md` (25+ KB) +**Executive Summary for Stakeholders** + +Contains: +- Problem/solution +- 3 patterns (Sequential, Await-Task-Complete, Retry) +- 3 entry points (CLI, API, Legacy) +- Phases 1-5 (52-58 hours) +- KMSvc questions (Q1-Q6) +- Risks & mitigations +- Success criteria +- Approval checklist + +**When to use**: Stakeholder review, getting buy-in, architecture approval + +--- + +## 🎯 THE ARCHITECTURE AT A GLANCE + +``` +User Message: "Analyze repo for security and quality every day at 2 AM" + ↓ +[llm-router Activity] + Reads: ActivityKnowledgeBase.json + Uses LLM to understand intent + Decides: Clone β†’ AnalyzeCode β†’ SecurityScan β†’ Combine β†’ Notify + Decides timeouts, retries from knowledge base + Detects schedule: "0 2 * * *" + Generates: CronWorkflowSpec + ↓ +[RoutingWorkflow] (Generic Executor) + Registers with Temporal cron: "0 2 * * *" + Every day at 2 AM: + 1. Clone repo + 2. Analyze code (timeout 10m, retry 3x if flaky) + 3. Security scan (timeout 15m, retry 2x) + 4. Combine results + 5. Send notification + Tracks each execution + ↓ +[Results] + Full execution history + Can check status anytime +``` + +--- + +## ✨ KEY FEATURES + +| Feature | Status | Docs | Tasks | +|---------|--------|------|-------| +| One-time workflows | βœ… | ROUTING_WORKFLOW_SPEC.md | 2.1-2.4, 3.x, 4.x | +| Scheduled workflows (cron) | βœ… | CRON_JOBS_QUICK_REFERENCE.md | 2.5, 3.4, 5.x | +| Intelligent routing (LLM) | βœ… | ROUTING_WORKFLOW_SPEC.md Part 2 | 2.x | +| Smart timeouts | βœ… | ROUTING_WORKFLOW_SPEC.md | 1.2, 2.4 | +| Smart retries | βœ… | ROUTING_WORKFLOW_SPEC.md | 1.2, 2.4 | +| Error handling | βœ… | ROUTING_WORKFLOW_SPEC.md | 3.4 | +| Parameter chaining | βœ… | ROUTING_WORKFLOW_SPEC.md | 2.1 | +| Temporal durability | βœ… | ROUTING_WORKFLOW_SPEC.md | 3.4 | +| HTTP API | βœ… | ROUTING_WORKFLOW_SPEC.md | 4.1-4.4 | +| CLI | βœ… | CRON_JOBS_QUICK_REFERENCE.md | 4.2 | +| Execution tracking | βœ… | CRON_JOBS_QUICK_REFERENCE.md | 5.x | + +--- + +## πŸ“‹ QUICK START FOR IMPLEMENTATION + +### Week 1: Foundation + LLM-Router + +``` +Day 1-2 (Mon-Tue): + Task 1.1: Go types (2h) + Task 1.2: Knowledge base JSON (3h) + Task 1.3: KB loader (2h) + Task 1.4: Validator (3h) + β†’ Deliverable: Core data structures working + +Day 3-5 (Wed-Fri): + Task 2.1: JSONPath resolver (3h) + Task 2.2: Activity skeleton (2h) + Task 2.3: LLM intent analysis (5h) + Task 2.4: Spec builder (4h) + Task 2.5: Cron builder (2h) + β†’ Deliverable: llm-router generates valid specs +``` + +### Week 2: RoutingWorkflow + API/CLI + +``` +Day 1-3 (Mon-Wed): + Task 3.1-3.6: RoutingWorkflow & executors (15-18h) + Task 3.5: Register in worker + β†’ Deliverable: Workflows execute, can submit via API + +Day 4-5 (Thu-Fri): + Task 4.1: API handlers (4h) + Task 4.2: CLI commands (5h) + Task 4.3: Server bootstrap (2h) + Task 4.4: Validation (2h) + β†’ Deliverable: Full HTTP API + CLI working +``` + +### Week 3: Testing + Documentation + +``` +Day 1-3 (Mon-Wed): + Task 5.1-5.4: All tests (8-12h) + β†’ Deliverable: >90% coverage, all tests pass + +Day 4-5 (Thu-Fri): + Task 6.1-6.4: Documentation (5-8h) + β†’ Deliverable: Complete docs, ready to ship +``` + +--- + +## πŸš€ HOW TO START TODAY + +### Step 1: Read & Understand (1-2 hours) +1. Read `ROUTING_WORKFLOW_SPEC.md` (main spec) +2. Read `IMPLEMENTATION_TASKS.md` (what to build) +3. Scan `CRON_JOBS_QUICK_REFERENCE.md` (understand cron) + +### Step 2: Assign Tasks +1. Engineer 1: Tasks 1.1-1.4, 2.1-2.5, 3.1-3.6 +2. Engineer 2: Tasks 4.1-4.4, 5.1-5.4, 6.1-6.4 + +### Step 3: Start Building +1. Begin with Task 1.1 (types.go) +2. Follow dependency order +3. Daily sync on blockers + +### Step 4: Gate Each Phase +- Phase 1 done? β†’ Start Phase 2 +- Phase 2 done? β†’ Start Phase 3 +- etc. + +--- + +## πŸ“Š EFFORT SUMMARY + +| Phase | Hours | Duration | Parallel | +|-------|-------|----------|----------| +| Phase 1: Foundation | 8-10 | Mon-Tue | No | +| Phase 2: LLM-Router | 12-15 | Wed-Fri + Mon | No | +| Phase 3: RoutingWorkflow | 15-18 | Tue-Thu | Can overlap w/ Phase 4 | +| Phase 4: API/CLI | 12-15 | Fri-Tue | Can overlap w/ Phase 3 | +| Phase 5: Testing | 8-12 | Wed-Fri | Sequential | +| Phase 6: Docs | 5-8 | Fri-Mon | Parallel w/ Phase 5 | +| **TOTAL** | **60-70** | **3-4 weeks** | **2 engineers** | + +--- + +## βœ… SUCCESS CRITERIA + +**Phase 1 Complete**: +- All types compile +- Knowledge base loads +- Validator catches errors +- All unit tests pass + +**Phase 2 Complete**: +- llm-router generates valid specs +- JSONPath resolution works +- Cron detection works +- Integration tests pass + +**Phase 3 Complete**: +- RoutingWorkflow executes any spec +- Error handling works +- State machine flow correct +- Registered in worker + +**Phase 4 Complete**: +- HTTP API working (all endpoints) +- CLI working (all commands) +- Input validation +- Can submit and check status + +**Phase 5 Complete**: +- >90% code coverage +- All scenarios pass +- Performance targets met +- No flaky tests + +**Phase 6 Complete**: +- API documentation complete +- CLI documentation complete +- Deployment guide complete +- User guide with examples + +--- + +## πŸ”— FILE LOCATIONS + +``` +Core Specification: + ~/workplace/Poimen/workflows/ROUTING_WORKFLOW_SPEC.md + +Task Breakdown: + ~/workplace/Poimen/workflows/IMPLEMENTATION_TASKS.md + +Cron Reference: + ~/workplace/Poimen/workflows/CRON_JOBS_QUICK_REFERENCE.md + +Stakeholder Review: + ~/workplace/Poimen/workflows/DESIGN_MASTER_REVIEW.md + +This README: + ~/workplace/Poimen/workflows/README_IMPLEMENTATION.md +``` + +--- + +## πŸ’‘ TIPS FOR SUCCESS + +1. **Start with types** (Task 1.1) + - Everything depends on these + - Make them flexible + - Good JSON schema helps + +2. **Knowledge base is critical** (Task 1.2) + - LLM decisions are based on this + - Make it comprehensive + - Document each activity well + +3. **Test llm-router early** (Task 2.3) + - This is highest risk + - Test with real LLM calls + - Validate output quality + +4. **RoutingWorkflow is the heart** (Task 3.4) + - Make sure state machine is solid + - Test error paths thoroughly + - Performance matters + +5. **API/CLI can be simple** (Tasks 4.x) + - Just thin wrappers + - Focus on DX (developer experience) + - Good error messages + +6. **Test everything** (Phase 5) + - Unit tests catch bugs early + - Integration tests find edge cases + - E2E tests validate full flow + - Load tests validate performance + +--- + +## πŸŽ“ LEARNING RESOURCES + +- **Temporal**: https://docs.temporal.io/ +- **Cron syntax**: https://crontab.guru/ +- **JSONPath**: https://goessner.net/articles/JsonPath/ +- **Go workflow patterns**: https://golang.org/pkg/workflow + +--- + +## πŸ“ž DECISION MAKER'S CHECKLIST + +Before starting implementation: + +- [ ] Do we have LLM access? (for llm-router) +- [ ] Is Temporal deployed? (task queue "poimen-taskqueue") +- [ ] Are activities registered? (CloneRepoActivity, etc) +- [ ] Do we have memory service? (for LLM calls) +- [ ] Team aligned on architecture? +- [ ] Timeline acceptable? (3-4 weeks) +- [ ] Resources allocated? (2 engineers) + +All yes? β†’ Ready to build! πŸš€ + +--- + +**This is a complete, implementable specification.** +Start with Phase 1, Task 1.1 today! + diff --git a/ROUTING_WORKFLOW_SPEC.md b/ROUTING_WORKFLOW_SPEC.md new file mode 100644 index 0000000..23a66ae --- /dev/null +++ b/ROUTING_WORKFLOW_SPEC.md @@ -0,0 +1,1198 @@ +# Poimen Routing Workflow Specification + +**Status**: Complete Standard +**Version**: 1.0 +**Architecture**: User Message β†’ LLM-Router β†’ JSON WorkflowSpec β†’ RoutingWorkflow + +--- + +## OVERVIEW + +### The Flow + +``` +User Message: "Analyze this GitHub repo for security and quality" + ↓ +[llm-router Activity] ← LLM-powered intelligent agent + Reads: ActivityKnowledgeBase.json + ↓ + Uses LLM to understand intent: + - What is user asking for? + - Which activities are needed? + - What order? + - What timeout for each? + - What retry strategy? + ↓ + Generates JSON WorkflowSpec + { + "name": "repo-analysis", + "input": {"repo": "https://...", "branch": "main"}, + "states": [ + { + "name": "Clone", + "type": "Task", + "resource": "CloneRepoActivity", + "parameters": {"repo": "${input.repo}"}, + "timeout": "5m", ← LLM decided + "retry": {...}, ← LLM decided + "next": "Analyze" + }, + {...more states...} + ] + } + ↓ +[RoutingWorkflow] ← Generic executor + Takes the JSON spec from llm-router + Executes states in order + Respects timeout/retry + ↓ +Final Results +``` + +### Key Insight + +**Intelligence is distributed**: +- **ActivityKnowledgeBase.json**: What activities exist, their constraints, capabilities +- **llm-router Activity**: Uses LLM to reason about user intent + pick activities + decide settings +- **RoutingWorkflow**: Dumb executor that just runs what the LLM decided + +--- + +## PART 1: ACTIVITY KNOWLEDGE BASE + +### What is it? + +Metadata about all available activities. The llm-router reads this to make intelligent decisions. + +### File Location + +``` +internal/routing/activity_knowledge_base.json +``` + +### Example Format + +```json +{ + "activities": [ + { + "name": "CloneRepoActivity", + "description": "Clone a Git repository to local filesystem", + "category": "source-control", + "inputs": { + "repo": { + "type": "string", + "description": "Repository URL (https://...)", + "required": true + }, + "branch": { + "type": "string", + "description": "Branch name to clone", + "required": false, + "default": "main" + } + }, + "outputs": { + "path": { + "type": "string", + "description": "Local filesystem path to cloned repo" + }, + "commit": { + "type": "string", + "description": "Current commit hash" + } + }, + "constraints": { + "defaultTimeout": "5m", + "isFlaky": false, + "recommendedRetries": 2, + "retryBackoff": 1.5, + "dependencies": [], + "notes": "Fast operation, rarely fails" + } + }, + { + "name": "AnalyzeCodeActivity", + "description": "Analyze code for quality issues, complexity, patterns", + "category": "analysis", + "inputs": { + "path": { + "type": "string", + "description": "Path to code directory", + "required": true + }, + "language": { + "type": "string", + "description": "Programming language (go, python, js, etc)", + "required": false, + "default": "auto-detect" + } + }, + "outputs": { + "quality": { + "type": "number", + "description": "Quality score 0-1" + }, + "issues": { + "type": "array", + "description": "List of issues found" + }, + "complexity": { + "type": "number", + "description": "Cyclomatic complexity" + } + }, + "constraints": { + "defaultTimeout": "10m", + "isFlaky": true, + "recommendedRetries": 3, + "retryBackoff": 2.0, + "dependencies": ["CloneRepoActivity"], + "notes": "Can timeout on large codebases, retry recommended" + } + }, + { + "name": "SecurityScanActivity", + "description": "Scan code for vulnerabilities and security issues", + "category": "security", + "inputs": { + "path": { + "type": "string", + "description": "Path to scan", + "required": true + } + }, + "outputs": { + "vulnerabilities": { + "type": "array", + "description": "List of vulnerabilities found" + }, + "score": { + "type": "number", + "description": "Security score 0-1" + } + }, + "constraints": { + "defaultTimeout": "15m", + "isFlaky": false, + "recommendedRetries": 2, + "retryBackoff": 1.5, + "dependencies": ["CloneRepoActivity"], + "notes": "Longer runtime, can be parallelized with other scans" + } + }, + { + "name": "PerformanceAnalysisActivity", + "description": "Analyze runtime performance, bottlenecks, optimization opportunities", + "category": "analysis", + "inputs": { + "path": { + "type": "string", + "description": "Path to code", + "required": true + } + }, + "outputs": { + "performance": { + "type": "number", + "description": "Performance score 0-1" + }, + "bottlenecks": { + "type": "array", + "description": "Performance bottlenecks" + } + }, + "constraints": { + "defaultTimeout": "10m", + "isFlaky": true, + "recommendedRetries": 2, + "retryBackoff": 2.0, + "dependencies": ["CloneRepoActivity"], + "notes": "Can timeout on complex analysis" + } + }, + { + "name": "JudgeActivity", + "description": "Make final judgment/recommendation based on analysis results", + "category": "judgment", + "inputs": { + "quality": { + "type": "object", + "description": "Code quality analysis result", + "required": true + }, + "security": { + "type": "object", + "description": "Security scan result", + "required": false + }, + "performance": { + "type": "object", + "description": "Performance analysis result", + "required": false + } + }, + "outputs": { + "verdict": { + "type": "string", + "description": "approved, needs-changes, rejected" + }, + "score": { + "type": "number", + "description": "Overall score" + }, + "recommendations": { + "type": "array", + "description": "Recommendations for improvement" + } + }, + "constraints": { + "defaultTimeout": "5m", + "isFlaky": false, + "recommendedRetries": 1, + "retryBackoff": 1.0, + "dependencies": ["AnalyzeCodeActivity"], + "notes": "Fast operation, only runs after analysis" + } + }, + { + "name": "CombineResultsActivity", + "description": "Combine multiple analysis results into final report", + "category": "aggregation", + "inputs": { + "results": { + "type": "array", + "description": "Array of previous step results", + "required": true + } + }, + "outputs": { + "report": { + "type": "object", + "description": "Final combined report" + } + }, + "constraints": { + "defaultTimeout": "2m", + "isFlaky": false, + "recommendedRetries": 1, + "retryBackoff": 1.0, + "dependencies": [], + "notes": "Fast, runs at the end" + } + }, + { + "name": "SendNotificationActivity", + "description": "Send notification with results to user", + "category": "notification", + "inputs": { + "result": { + "type": "object", + "description": "Results to send", + "required": true + }, + "channel": { + "type": "string", + "description": "Where to send (email, slack, etc)", + "required": false, + "default": "email" + } + }, + "outputs": { + "sent": { + "type": "boolean", + "description": "Was notification sent?" + } + }, + "constraints": { + "defaultTimeout": "2m", + "isFlaky": true, + "recommendedRetries": 3, + "retryBackoff": 2.0, + "dependencies": [], + "notes": "External dependency, retry recommended" + } + } + ] +} +``` + +### Knowledge Base Fields Explained + +| Field | Purpose | Used By | +|-------|---------|---------| +| **name** | Unique activity identifier | llm-router for routing | +| **description** | What the activity does | llm-router for reasoning | +| **category** | Type of activity (analysis, security, etc) | llm-router for grouping | +| **inputs** | What the activity accepts | llm-router to chain activities | +| **outputs** | What the activity produces | llm-router to connect outputs to next activity inputs | +| **constraints.defaultTimeout** | Recommended timeout | llm-router to decide timeout | +| **constraints.isFlaky** | Does it fail often? | llm-router to decide retry count | +| **constraints.recommendedRetries** | How many retries? | llm-router to configure retry | +| **constraints.retryBackoff** | Backoff multiplier | llm-router to configure exponential backoff | +| **constraints.dependencies** | Must run after X | llm-router to order activities | +| **constraints.notes** | Special handling notes | llm-router for reasoning | + +--- + +## PART 2: LLM-ROUTER ACTIVITY + +### What is it? + +An Activity that uses an LLM to intelligently decide: +1. Which activities are needed +2. What order to execute them +3. What timeout for each activity +4. What retry policy for each activity +5. How to chain parameters (JSONPath) + +### Input to llm-router + +```json +{ + "message": "Analyze this GitHub repo for security and code quality", + "repo": "https://github.com/rockliang/poimen", + "branch": "main" +} +``` + +### Output from llm-router + +```json +{ + "name": "repo-analysis-workflow", + "input": { + "repo": "https://github.com/rockliang/poimen", + "branch": "main" + }, + "states": [ + { + "name": "Clone", + "type": "Task", + "resource": "CloneRepoActivity", + "parameters": { + "repo": "${input.repo}", + "branch": "${input.branch}" + }, + "timeout": "5m", + "retry": { + "maxAttempts": 2, + "backoffRate": 1.5, + "initialInterval": "1s" + }, + "next": "Analyze" + }, + { + "name": "Analyze", + "type": "Task", + "resource": "AnalyzeCodeActivity", + "parameters": { + "path": "${Clone.output.path}" + }, + "timeout": "10m", + "retry": { + "maxAttempts": 3, + "backoffRate": 2.0, + "initialInterval": "1s" + }, + "catch": [ + { + "errorEquals": ["Timeout"], + "next": "HandleTimeout" + } + ], + "next": "SecurityScan" + }, + { + "name": "SecurityScan", + "type": "Task", + "resource": "SecurityScanActivity", + "parameters": { + "path": "${Clone.output.path}" + }, + "timeout": "15m", + "retry": { + "maxAttempts": 2, + "backoffRate": 1.5, + "initialInterval": "1s" + }, + "next": "Combine" + }, + { + "name": "Combine", + "type": "Task", + "resource": "CombineResultsActivity", + "parameters": { + "results": [ + "${Analyze.output}", + "${SecurityScan.output}" + ] + }, + "timeout": "2m", + "retry": { + "maxAttempts": 1, + "backoffRate": 1.0, + "initialInterval": "1s" + }, + "next": "SendResults" + }, + { + "name": "SendResults", + "type": "Task", + "resource": "SendNotificationActivity", + "parameters": { + "result": "${Combine.output.report}", + "channel": "email" + }, + "timeout": "2m", + "retry": { + "maxAttempts": 3, + "backoffRate": 2.0, + "initialInterval": "1s" + }, + "end": true + }, + { + "name": "HandleTimeout", + "type": "Fail", + "error": "AnalysisTimeout", + "cause": "Code analysis timed out after 10 minutes" + } + ] +} +``` + +### LLM-Router Logic (Pseudo-code) + +```python +def llm_router_activity(message: str, repo: str, branch: str) -> WorkflowSpec: + # 1. Load knowledge base + knowledge_base = load_activity_knowledge_base() + + # 2. Use LLM to understand intent + intent = llm.analyze_intent(message) + # intent = { + # "needs_security_scan": true, + # "needs_code_analysis": true, + # "needs_performance_analysis": false, + # "needs_final_judgment": true, + # } + + # 3. Select activities based on intent + selected_activities = [] + if intent.needs_security_scan: + selected_activities.append(knowledge_base["SecurityScanActivity"]) + if intent.needs_code_analysis: + selected_activities.append(knowledge_base["AnalyzeCodeActivity"]) + if intent.needs_performance_analysis: + selected_activities.append(knowledge_base["PerformanceAnalysisActivity"]) + + # 4. Order activities by dependencies + ordered = topological_sort_by_dependencies(selected_activities) + + # 5. Build workflow states with intelligent settings + states = [] + + # Always start with Clone + clone_state = { + "name": "Clone", + "type": "Task", + "resource": "CloneRepoActivity", + "parameters": { + "repo": "${input.repo}", + "branch": "${input.branch}" + }, + "timeout": "5m", # From knowledge base + "retry": { + "maxAttempts": 2, + "backoffRate": 1.5, + "initialInterval": "1s" + }, + "next": ordered[0]["name"] + } + states.append(clone_state) + + # Add selected activities in order + for i, activity in enumerate(ordered): + state = { + "name": activity["name"].replace("Activity", ""), + "type": "Task", + "resource": activity["name"], + "parameters": { + # Use LLM to map inputs + # For each input, decide: use output from previous step? use input param? use constant? + }, + "timeout": activity["constraints"]["defaultTimeout"], + "retry": { + "maxAttempts": activity["constraints"]["recommendedRetries"], + "backoffRate": activity["constraints"]["retryBackoff"], + "initialInterval": "1s" + }, + "next": ordered[i+1]["name"] if i+1 < len(ordered) else "SendResults" + } + + # Add error handling if flaky + if activity["constraints"]["isFlaky"]: + state["catch"] = [ + { + "errorEquals": ["Timeout"], + "next": "HandleTimeout" + } + ] + + states.append(state) + + # Add final notification + states.append({ + "name": "SendResults", + "type": "Task", + "resource": "SendNotificationActivity", + "parameters": { + "result": "${" + ordered[-1]["name"].replace("Activity", "") + ".output}", + "channel": "email" + }, + "timeout": "2m", + "retry": {"maxAttempts": 3, "backoffRate": 2.0, "initialInterval": "1s"}, + "end": true + }) + + # Add error handler + states.append({ + "name": "HandleTimeout", + "type": "Fail", + "error": "AnalysisTimeout", + "cause": "Activity timed out" + }) + + # 6. Return complete workflow spec + return WorkflowSpec( + name="generated-workflow", + input={"repo": repo, "branch": branch}, + states=states + ) +``` + +### Go Implementation Skeleton + +```go +// cmd/worker/activities/llm_router.go + +type LLMRouterInput struct { + Message string + Repo string + Branch string +} + +func LLMRouterActivity(ctx context.Context, input LLMRouterInput) (routing.WorkflowSpec, error) { + // 1. Load activity knowledge base + kb, err := LoadActivityKnowledgeBase() + if err != nil { + return routing.WorkflowSpec{}, err + } + + // 2. Use LLM to understand intent + // Call memory service to get LLM analysis + intent, err := llm.AnalyzeIntent(ctx, input.Message, kb) + if err != nil { + return routing.WorkflowSpec{}, err + } + + // 3. Build workflow spec intelligently + spec := buildWorkflowSpec(intent, input, kb) + + return spec, nil +} + +func buildWorkflowSpec(intent Intent, input LLMRouterInput, kb ActivityKnowledgeBase) routing.WorkflowSpec { + // Create workflow with intelligent settings from knowledge base + // ... +} +``` + +--- + +## PART 3: ROUTING WORKFLOW (The Executor) + +### What is it? + +A generic Temporal workflow that executes ANY WorkflowSpec generated by llm-router. + +### Input + +```go +type RoutingWorkflowInput struct { + WorkflowSpec routing.WorkflowSpec // Generated by llm-router +} +``` + +### Implementation + +```go +func RoutingWorkflow(ctx workflow.Context, spec routing.WorkflowSpec) (Result, error) { + logger := workflow.GetLogger(ctx) + + // Initialize context + ec := &ExecutionContext{ + Input: spec.Input, + StepResults: make(map[string]interface{}), + } + + // Find first state + currentStateName := FindFirstState(spec.States) + + // State machine loop + for { + logger.Info("Executing state", "state", currentStateName) + + state := FindStateByName(spec.States, currentStateName) + if state == nil { + return Result{}, fmt.Errorf("state not found: %s", currentStateName) + } + + // Execute state (Task, Pass, Fail, etc) + result, err := ExecuteState(ctx, state, ec) + + // Error handling + if err != nil { + for _, catchClause := range state.Catch { + if MatchesError(err, catchClause.ErrorEquals) { + logger.Info("Error caught", "handler", catchClause.Next) + currentStateName = catchClause.Next + break + } + } + return Result{}, fmt.Errorf("state %s failed: %w", currentStateName, err) + } + + // Store result + ec.StepResults[state.Name] = result + + // Check if done + if state.End { + logger.Info("Workflow completed") + return Result{ + FinalOutput: result, + Status: "COMPLETED", + }, nil + } + + // Move to next + currentStateName = state.Next + } +} +``` + +--- + +## PART 4: WORKFLOW SPEC FORMAT + +### State Types + +The llm-router generates states with these types: + +#### Task (Execute Activity) + +```json +{ + "name": "Clone", + "type": "Task", + "resource": "CloneRepoActivity", + "parameters": { + "repo": "${input.repo}", + "branch": "${input.branch:main}" + }, + "timeout": "5m", + "retry": { + "maxAttempts": 3, + "backoffRate": 2.0, + "initialInterval": "1s" + }, + "catch": [ + { + "errorEquals": ["Timeout", "NetworkError"], + "next": "HandleError" + } + ], + "next": "Analyze" +} +``` + +**LLM decides**: +- Which activity to use +- How to chain parameters (${Clone.output.path}) +- Timeout (from knowledge base + LLM reasoning) +- Retry policy (from knowledge base + isFlaky flag) +- Error handling (catch blocks for flaky activities) + +#### Pass (Static Output) + +```json +{ + "name": "CombineResults", + "type": "Pass", + "result": { + "status": "success", + "report": "Analysis complete" + }, + "next": "SendResults" +} +``` + +#### Fail (Terminate) + +```json +{ + "name": "HandleError", + "type": "Fail", + "error": "AnalysisFailed", + "cause": "Security scan timed out" +} +``` + +--- + +## PART 5: EXECUTION FLOW EXAMPLE + +### User Request + +``` +"Analyze the GitHub repo rockliang/poimen for security vulnerabilities and code quality" +``` + +### Step 1: llm-router Activity + +``` +Input: {message: "...", repo: "https://github.com/rockliang/poimen", branch: "main"} + +LLM Analysis: + - User wants: security analysis + code quality + - Sequence: Clone β†’ Analyze (quality) β†’ SecurityScan β†’ CombineResults β†’ Notify + - Timeouts: Use defaults from knowledge base + - Clone: 5m (not flaky) + - Analyze: 10m (flaky, so 3 retries) + - SecurityScan: 15m (not flaky, so 2 retries) + - Combine: 2m (not flaky, 1 retry) + - Error handling: Add catch blocks for flaky activities (Analyze) + +Output: WorkflowSpec (as JSON above) +``` + +### Step 2: RoutingWorkflow Executes + +``` +State 1: Clone + Execute: CloneRepoActivity({repo: "https://...", branch: "main"}) + Timeout: 5m (from llm decision) + Result: {path: "/tmp/cloned-repo"} + ↓ +State 2: Analyze + Execute: AnalyzeCodeActivity({path: "/tmp/cloned-repo"}) + Timeout: 10m + Retry: 3 attempts (from llm decision, isFlaky=true) + Attempt 1: TIMEOUT β†’ Wait 1s + Attempt 2: TIMEOUT β†’ Wait 2s + Attempt 3: SUCCESS β†’ {quality: 0.85, issues: [...]} + ↓ +State 3: SecurityScan + Execute: SecurityScanActivity({path: "/tmp/cloned-repo"}) + Timeout: 15m + Retry: 2 attempts (from llm decision, isFlaky=false) + Result: {vulnerabilities: [], score: 0.95} + ↓ +State 4: CombineResults + Execute: CombineResultsActivity({results: [...]}) + Result: {report: {...}} + ↓ +State 5: SendResults + Execute: SendNotificationActivity({result: {...}, channel: "email"}) + Result: {sent: true} + ↓ +βœ… WORKFLOW COMPLETED +``` + +--- + +## PART 6: GO TYPE DEFINITIONS + +```go +package routing + +import "time" + +// WorkflowSpec is generated by llm-router (one-time execution) +type WorkflowSpec struct { + Name string `json:"name"` + Input map[string]interface{} `json:"input"` + States []State `json:"states"` +} + +// CronWorkflowSpec is generated by llm-router (scheduled execution) +type CronWorkflowSpec struct { + Name string `json:"name"` + Type string `json:"type"` // "CronWorkflow" + Schedule string `json:"schedule"` // Cron expression (e.g., "0 2 * * *") + Timezone string `json:"timezone"` // "UTC", "America/New_York", etc + Input map[string]interface{} `json:"input"` // Fixed input for each run + States []State `json:"states"` // Workflow states + MaxConcurrent int `json:"maxConcurrent,omitempty"` // Max parallel runs (default 1) + Timeout string `json:"timeout,omitempty"` // Overall timeout per run + EnableHistory bool `json:"enableHistory,omitempty"` // Keep execution history +} + +// State is a step in the workflow +type State struct { + Name string + Type StateType // "Task", "Pass", "Fail" + + // Task fields + Resource string + Parameters map[string]interface{} + Timeout string + Retry *RetryPolicy + Catch []CatchClause + + // Pass fields + Result interface{} + + // Fail fields + Error string + Cause string + + // Transition + Next string + End bool +} + +type StateType string + +const ( + StateTypeTask StateType = "Task" + StateTypePass StateType = "Pass" + StateTypeFail StateType = "Fail" +) + +// RetryPolicy (LLM decides this from knowledge base) +type RetryPolicy struct { + MaxAttempts int32 `json:"maxAttempts"` + BackoffRate float64 `json:"backoffRate"` + InitialInterval string `json:"initialInterval"` + MaxInterval string `json:"maxInterval,omitempty"` +} + +// CatchClause for error handling +type CatchClause struct { + ErrorEquals []string `json:"errorEquals"` + Next string `json:"next"` +} + +// ExecutionContext tracks state +type ExecutionContext struct { + Input map[string]interface{} + StepResults map[string]interface{} +} + +// Result is final output +type Result struct { + FinalOutput interface{} + Status string + Error error +} +``` + +--- + +## PART 7: CRON JOB SCHEDULING + +### What is it? + +llm-router can generate workflows that execute **on a schedule** (cron jobs), not just one-time. + +### Example: Daily Security Scan + +User: "Scan all repositories for security vulnerabilities every day at 2 AM" + +llm-router generates: + +```json +{ + "name": "daily-security-scan", + "type": "CronWorkflow", + "schedule": "0 2 * * *", + "timezone": "UTC", + "input": { + "repos": [ + "https://github.com/rockliang/poimen", + "https://github.com/rockliang/workflow-engine" + ] + }, + "states": [ + { + "name": "Clone", + "type": "Task", + "resource": "CloneRepoActivity", + "parameters": { + "repo": "${input.repos[0]}" + }, + "timeout": "5m", + "retry": {"maxAttempts": 2, "backoffRate": 1.5, "initialInterval": "1s"}, + "next": "SecurityScan" + }, + { + "name": "SecurityScan", + "type": "Task", + "resource": "SecurityScanActivity", + "parameters": {"path": "${Clone.output.path}"}, + "timeout": "15m", + "retry": {"maxAttempts": 2, "backoffRate": 1.5, "initialInterval": "1s"}, + "next": "SendAlert" + }, + { + "name": "SendAlert", + "type": "Task", + "resource": "SendNotificationActivity", + "parameters": { + "result": "${SecurityScan.output}", + "channel": "slack" + }, + "timeout": "2m", + "retry": {"maxAttempts": 3, "backoffRate": 2.0, "initialInterval": "1s"}, + "end": true + } + ] +} +``` + +### Cron Format + +Standard Unix cron syntax: + +``` +β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€ minute (0 - 59) +β”‚ β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€ hour (0 - 23) +β”‚ β”‚ β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€ day of month (1 - 31) +β”‚ β”‚ β”‚ β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€ month (1 - 12) +β”‚ β”‚ β”‚ β”‚ β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€ day of week (0 - 6, 0 = Sunday) +β”‚ β”‚ β”‚ β”‚ β”‚ +β”‚ β”‚ β”‚ β”‚ β”‚ +* * * * * + +Examples: +0 2 * * * = Every day at 2:00 AM +0 */6 * * * = Every 6 hours +0 9 * * 1-5 = Weekdays at 9 AM +0 0 1 * * = First day of month at midnight +30 2 * * 0 = Every Sunday at 2:30 AM +*/15 * * * * = Every 15 minutes +0 12 * * * = Every day at noon +``` + +### Cron Workflow Fields + +```go +type CronWorkflowSpec struct { + Name string `json:"name"` + Type string `json:"type"` // "CronWorkflow" + Schedule string `json:"schedule"` // Cron expression + Timezone string `json:"timezone"` // "UTC", "America/New_York", etc + Input map[string]interface{} `json:"input"` // Fixed input for each run + States []State `json:"states"` // Same as regular workflow + + // Optional + MaxConcurrent int `json:"maxConcurrent,omitempty"` // Max parallel executions (default 1) + Timeout string `json:"timeout,omitempty"` // Overall timeout per run (default 1h) + EnableHistory bool `json:"enableHistory,omitempty"` // Keep execution history? +} +``` + +### Temporal Cron Job Implementation + +```go +func SubmitCronWorkflow(cronSpec CronWorkflowSpec) error { + client, _ := client.Dial(client.ClientOptions{ + HostPort: "temporal-frontend.temporal:7233", + }) + + // For cron, we still use RoutingWorkflow + // But Temporal handles the scheduling + + _, err := client.ExecuteWorkflow( + context.Background(), + client.StartWorkflowOptions{ + ID: cronSpec.Name, + TaskQueue: "poimen-taskqueue", + CronSchedule: cronSpec.Schedule, // ← Temporal cron support + }, + RoutingWorkflow, + cronSpec.States, + ) + + return err +} +``` + +### Example Use Cases + +``` +1. Daily Security Scan (0 2 * * *) + Input: {repos: [...]} + Runs: Every day at 2 AM + ↓ + Clone all repos β†’ Security scan β†’ Send alert + +2. Hourly Code Quality Check (0 * * * *) + Input: {repo: "https://github.com/rockliang/poimen"} + Runs: Every hour + ↓ + Clone β†’ Analyze β†’ Judge β†’ Log results + +3. Weekly Performance Benchmark (0 3 * * 0) + Input: {repos: [...], benchmark: true} + Runs: Every Sunday at 3 AM + ↓ + Clone β†’ Performance test β†’ Compare with baseline β†’ Report + +4. Every 15 minutes Health Check (*/15 * * * *) + Input: {service: "api"} + Runs: Every 15 minutes + ↓ + Check service health β†’ Send metrics β†’ Alert if down +``` + +### llm-router for Cron Jobs + +User: "Run daily security scan at 2 AM UTC on all our repositories" + +llm-router decision: + +```python +def llm_router_activity(message: str) -> WorkflowSpec: + intent = llm.analyze_intent(message) + # intent = { + # "is_scheduled": True, + # "schedule": "0 2 * * *", # LLM extracts from "2 AM" + # "timezone": "UTC", + # "needs_security_scan": True, + # "target": "all repositories" + # } + + if intent.is_scheduled: + return CronWorkflowSpec( + name="daily-security-scan", + type="CronWorkflow", + schedule=intent.schedule, # "0 2 * * *" + timezone=intent.timezone, # "UTC" + input={"repos": get_all_repos()}, + states=[ + # Same states as one-time workflow + ] + ) + else: + return OneTimeWorkflowSpec(...) +``` + +### Execution History + +When `enableHistory=true`, Temporal tracks all cron executions: + +``` +Cron Workflow ID: daily-security-scan + +Execution 1: 2025-02-01 02:00:00 UTC β†’ COMPLETED +Execution 2: 2025-02-02 02:00:00 UTC β†’ COMPLETED +Execution 3: 2025-02-03 02:00:00 UTC β†’ FAILED (timeout) +Execution 4: 2025-02-04 02:00:00 UTC β†’ COMPLETED +Execution 5: 2025-02-05 02:00:00 UTC β†’ COMPLETED (in progress...) +``` + +--- + +## PART 8: VALIDATION + +Before RoutingWorkflow executes the spec, validate: + +``` +βœ… All state names are unique +βœ… All state.next references exist +βœ… All catch.next references exist +βœ… All Task states have resource defined +βœ… No circular loops (Aβ†’Bβ†’Cβ†’A) +βœ… At least one path leads to end +βœ… No dead ends +βœ… Timeout format is valid (e.g., "5m") +βœ… Retry settings are valid +βœ… JSONPath expressions are syntactically valid +``` + +--- + +## SUMMARY + +### The Intelligence is in llm-router + +**llm-router activity**: +- βœ… Reads ActivityKnowledgeBase.json +- βœ… Uses LLM to understand user intent +- βœ… Intelligently selects activities to run +- βœ… Orders them by dependencies +- βœ… Decides timeout for EACH activity (from knowledge base + LLM reasoning) +- βœ… Decides retry policy for EACH activity (flaky? retry 3x : 1x) +- βœ… Chains parameters using JSONPath (${Clone.output.path}) +- βœ… Adds error handling (catch blocks) +- βœ… **Detects scheduled workflows** (cron jobs) and generates CronWorkflowSpec +- βœ… Returns complete JSON WorkflowSpec (one-time or recurring) + +### The Executor is Generic + +**RoutingWorkflow**: +- βœ… Takes JSON spec from llm-router +- βœ… Executes states in order +- βœ… Respects timeout/retry +- βœ… Handles errors with catch blocks +- βœ… Works for both one-time and cron workflows +- βœ… Returns results + +### Execution Modes + +| Mode | Triggered By | Execution | Example | +|------|---|---|---| +| **One-Time** | API call / CLI / Direct | Runs once, returns immediately | Analyze repo, generate report | +| **Cron Job** | Schedule | Runs repeatedly on schedule | Daily security scan at 2 AM | +| **Blocking** | User waits | Synchronous, returns results | Execute template, get output | +| **Async** | User polls | Asynchronous, check status later | Submit workflow, poll status | + +### Files Needed + +``` +1. internal/routing/activity_knowledge_base.json + ↑ Static metadata about activities (timeouts, retry, flaky, etc) + +2. cmd/worker/activities/llm_router.go + ↑ LLM-powered activity that generates WorkflowSpec or CronWorkflowSpec + +3. statemachine/routing_workflow.go + ↑ Generic executor for any spec (one-time or cron) + +4. internal/routing/validator.go + ↑ Validate specs before execution + ↑ Validate cron expressions + +5. cmd/api-server/cron_handlers.go (optional) + ↑ Endpoints to manage cron workflows + ↑ GET /api/v1/cron/{id}/status + ↑ DELETE /api/v1/cron/{id} (cancel) +``` + +This is the **correct, complete architecture**! 🎯 + +### Quick Feature Checklist + +- βœ… **One-time workflows** (API/CLI submit) +- βœ… **Scheduled workflows** (Cron jobs with schedule) +- βœ… **Intelligent routing** (LLM decides what to run) +- βœ… **Smart timeouts** (From activity knowledge base) +- βœ… **Smart retries** (Based on isFlaky flag) +- βœ… **Error handling** (Catch blocks for flaky activities) +- βœ… **Parameter chaining** (JSONPath resolution) +- βœ… **Execution history** (For cron jobs) +- βœ… **Temporal durability** (Automatic replay) + diff --git a/internal/routing/types.go b/internal/routing/types.go new file mode 100644 index 0000000..b623c39 --- /dev/null +++ b/internal/routing/types.go @@ -0,0 +1,152 @@ +package routing + +import "time" + +// WorkflowSpec is generated by llm-router (one-time execution) +type WorkflowSpec struct { + Name string `json:"name"` + Input map[string]interface{} `json:"input"` + States []State `json:"states"` +} + +// CronWorkflowSpec is generated by llm-router (scheduled execution) +type CronWorkflowSpec struct { + Name string `json:"name"` + Type string `json:"type"` // "CronWorkflow" + Schedule string `json:"schedule"` // Cron expression (e.g., "0 2 * * *") + Timezone string `json:"timezone"` // "UTC", "America/New_York", etc + Input map[string]interface{} `json:"input"` // Fixed input for each run + States []State `json:"states"` // Workflow states + MaxConcurrent int `json:"maxConcurrent,omitempty"` // Max parallel runs (default 1) + Timeout string `json:"timeout,omitempty"` // Overall timeout per run + EnableHistory bool `json:"enableHistory,omitempty"` // Keep execution history +} + +// State is a step in the workflow +type State struct { + Name string + Type StateType `json:"type"` + + // Task fields + Resource string `json:"resource,omitempty"` + Parameters map[string]interface{} `json:"parameters,omitempty"` + Timeout string `json:"timeout,omitempty"` + Retry *RetryPolicy `json:"retry,omitempty"` + Catch []CatchClause `json:"catch,omitempty"` + + // Pass fields + Result interface{} `json:"result,omitempty"` + + // Fail fields + Error string `json:"error,omitempty"` + Cause string `json:"cause,omitempty"` + + // Transition + Next string `json:"next,omitempty"` + End bool `json:"end,omitempty"` +} + +// StateType defines valid state types +type StateType string + +const ( + StateTypeTask StateType = "Task" + StateTypePass StateType = "Pass" + StateTypeFail StateType = "Fail" +) + +// RetryPolicy defines retry behavior for activities +type RetryPolicy struct { + MaxAttempts int32 `json:"maxAttempts"` + BackoffRate float64 `json:"backoffRate"` + InitialInterval string `json:"initialInterval"` + MaxInterval string `json:"maxInterval,omitempty"` +} + +// CatchClause defines error handling +type CatchClause struct { + ErrorEquals []string `json:"errorEquals"` + ResultPath *string `json:"resultPath,omitempty"` + Next string `json:"next"` +} + +// ExecutionContext tracks state during workflow execution +type ExecutionContext struct { + Input map[string]interface{} + StepResults map[string]interface{} + CurrentState string + History []ExecutionEvent +} + +// ExecutionEvent tracks individual state execution +type ExecutionEvent struct { + Timestamp time.Time + State string + Type string // "Started", "Completed", "Failed", "Retried" + Result interface{} + Error error +} + +// PollParams for AwaitTaskComplete states +type PollParams struct { + QueueName string + CorrelationID string + PollInterval time.Duration + Timeout time.Duration +} + +// PollResult is the result of polling +type PollResult struct { + Result interface{} + Status string +} + +// Result is the final workflow output +type Result struct { + FinalOutput interface{} + Status string // "COMPLETED", "FAILED" + Error error +} + +// Heartbeat contains state for polling activities +type Heartbeat struct { + CorrelationID string + Queue string + Attempt int + Elapsed time.Duration + LastCheck time.Time +} + +// ActivityMetadata describes an activity's capabilities and constraints +type ActivityMetadata struct { + Name string + Description string + Category string + Inputs map[string]InputField + Outputs map[string]OutputField + Constraints Constraints +} + +// InputField describes an activity input parameter +type InputField struct { + Type string `json:"type"` + Description string `json:"description"` + Required bool `json:"required"` + Default interface{} `json:"default,omitempty"` +} + +// OutputField describes an activity output field +type OutputField struct { + Type string `json:"type"` + Description string `json:"description"` +} + +// Constraints describes activity execution constraints +type Constraints struct { + DefaultTimeout string + IsFlaky bool + RecommendedRetries int + RetryBackoff float64 + Dependencies []string + Notes string +} diff --git a/internal/routing/types_test.go b/internal/routing/types_test.go new file mode 100644 index 0000000..82a4b36 --- /dev/null +++ b/internal/routing/types_test.go @@ -0,0 +1,327 @@ +package routing + +import ( + "encoding/json" + "testing" +) + +func TestWorkflowSpecMarshal(t *testing.T) { + spec := WorkflowSpec{ + Name: "test-workflow", + Input: map[string]interface{}{ + "repo": "https://github.com/test/repo", + }, + States: []State{ + { + Name: "Clone", + Type: StateTypeTask, + Resource: "CloneRepoActivity", + Parameters: map[string]interface{}{ + "repo": "${input.repo}", + }, + Timeout: "5m", + Next: "Analyze", + }, + }, + } + + // Marshal to JSON + data, err := json.Marshal(spec) + if err != nil { + t.Fatalf("Failed to marshal: %v", err) + } + + // Unmarshal back + var spec2 WorkflowSpec + err = json.Unmarshal(data, &spec2) + if err != nil { + t.Fatalf("Failed to unmarshal: %v", err) + } + + // Verify + if spec2.Name != spec.Name { + t.Errorf("Name mismatch: %s != %s", spec2.Name, spec.Name) + } + if len(spec2.States) != len(spec.States) { + t.Errorf("State count mismatch: %d != %d", len(spec2.States), len(spec.States)) + } +} + +func TestCronWorkflowSpecMarshal(t *testing.T) { + spec := CronWorkflowSpec{ + Name: "daily-scan", + Type: "CronWorkflow", + Schedule: "0 2 * * *", + Timezone: "UTC", + MaxConcurrent: 1, + EnableHistory: true, + Input: map[string]interface{}{ + "repos": []string{"repo1", "repo2"}, + }, + } + + // Marshal to JSON + data, err := json.Marshal(spec) + if err != nil { + t.Fatalf("Failed to marshal: %v", err) + } + + // Unmarshal back + var spec2 CronWorkflowSpec + err = json.Unmarshal(data, &spec2) + if err != nil { + t.Fatalf("Failed to unmarshal: %v", err) + } + + // Verify + if spec2.Schedule != spec.Schedule { + t.Errorf("Schedule mismatch: %s != %s", spec2.Schedule, spec.Schedule) + } + if spec2.Timezone != spec.Timezone { + t.Errorf("Timezone mismatch: %s != %s", spec2.Timezone, spec.Timezone) + } + if spec2.EnableHistory != spec.EnableHistory { + t.Errorf("EnableHistory mismatch: %v != %v", spec2.EnableHistory, spec.EnableHistory) + } +} + +func TestRetryPolicyMarshal(t *testing.T) { + policy := RetryPolicy{ + MaxAttempts: 3, + BackoffRate: 2.0, + InitialInterval: "1s", + MaxInterval: "1m", + } + + data, err := json.Marshal(policy) + if err != nil { + t.Fatalf("Failed to marshal: %v", err) + } + + var policy2 RetryPolicy + err = json.Unmarshal(data, &policy2) + if err != nil { + t.Fatalf("Failed to unmarshal: %v", err) + } + + if policy2.MaxAttempts != policy.MaxAttempts { + t.Errorf("MaxAttempts mismatch: %d != %d", policy2.MaxAttempts, policy.MaxAttempts) + } + if policy2.BackoffRate != policy.BackoffRate { + t.Errorf("BackoffRate mismatch: %f != %f", policy2.BackoffRate, policy.BackoffRate) + } +} + +func TestStateMarshal(t *testing.T) { + state := State{ + Name: "Analyze", + Type: StateTypeTask, + Resource: "AnalyzeCodeActivity", + Parameters: map[string]interface{}{ + "path": "${Clone.output.path}", + }, + Timeout: "10m", + Retry: &RetryPolicy{ + MaxAttempts: 3, + BackoffRate: 2.0, + InitialInterval: "1s", + }, + Catch: []CatchClause{ + { + ErrorEquals: []string{"Timeout"}, + Next: "HandleTimeout", + }, + }, + Next: "Judge", + } + + data, err := json.Marshal(state) + if err != nil { + t.Fatalf("Failed to marshal: %v", err) + } + + var state2 State + err = json.Unmarshal(data, &state2) + if err != nil { + t.Fatalf("Failed to unmarshal: %v", err) + } + + if state2.Name != state.Name { + t.Errorf("Name mismatch: %s != %s", state2.Name, state.Name) + } + if state2.Type != state.Type { + t.Errorf("Type mismatch: %s != %s", state2.Type, state.Type) + } + if len(state2.Catch) != len(state.Catch) { + t.Errorf("Catch count mismatch: %d != %d", len(state2.Catch), len(state.Catch)) + } +} + +func TestPassState(t *testing.T) { + state := State{ + Name: "SetSuccess", + Type: StateTypePass, + Result: map[string]interface{}{"status": "success"}, + End: true, + } + + data, err := json.Marshal(state) + if err != nil { + t.Fatalf("Failed to marshal: %v", err) + } + + var state2 State + err = json.Unmarshal(data, &state2) + if err != nil { + t.Fatalf("Failed to unmarshal: %v", err) + } + + if state2.Type != StateTypePass { + t.Errorf("Type should be Pass, got: %s", state2.Type) + } + if !state2.End { + t.Error("End should be true") + } +} + +func TestFailState(t *testing.T) { + state := State{ + Name: "HandleError", + Type: StateTypeFail, + Error: "InvalidInput", + Cause: "Repository URL is invalid", + } + + data, err := json.Marshal(state) + if err != nil { + t.Fatalf("Failed to marshal: %v", err) + } + + var state2 State + err = json.Unmarshal(data, &state2) + if err != nil { + t.Fatalf("Failed to unmarshal: %v", err) + } + + if state2.Type != StateTypeFail { + t.Errorf("Type should be Fail, got: %s", state2.Type) + } + if state2.Error != "InvalidInput" { + t.Errorf("Error mismatch: %s != InvalidInput", state2.Error) + } +} + +func TestExecutionContextInit(t *testing.T) { + ec := &ExecutionContext{ + Input: map[string]interface{}{"repo": "test"}, + StepResults: make(map[string]interface{}), + History: make([]ExecutionEvent, 0), + } + + if ec.Input == nil { + t.Error("Input should not be nil") + } + if ec.StepResults == nil { + t.Error("StepResults should not be nil") + } + if ec.History == nil { + t.Error("History should not be nil") + } +} + +func TestComplexWorkflowSpec(t *testing.T) { + // Test a realistic workflow spec + spec := WorkflowSpec{ + Name: "code-review", + Input: map[string]interface{}{ + "repo": "https://github.com/rockliang/poimen", + "branch": "feature/x", + }, + States: []State{ + { + Name: "Clone", + Type: StateTypeTask, + Resource: "CloneRepoActivity", + Parameters: map[string]interface{}{ + "repo": "${input.repo}", + "branch": "${input.branch}", + }, + Timeout: "5m", + Retry: &RetryPolicy{ + MaxAttempts: 2, + BackoffRate: 1.5, + InitialInterval: "1s", + }, + Next: "Analyze", + }, + { + Name: "Analyze", + Type: StateTypeTask, + Resource: "AnalyzeCodeActivity", + Parameters: map[string]interface{}{ + "path": "${Clone.output.path}", + }, + Timeout: "10m", + Retry: &RetryPolicy{ + MaxAttempts: 3, + BackoffRate: 2.0, + InitialInterval: "1s", + }, + Catch: []CatchClause{ + { + ErrorEquals: []string{"Timeout"}, + Next: "HandleTimeout", + }, + }, + Next: "Judge", + }, + { + Name: "Judge", + Type: StateTypeTask, + Resource: "JudgeActivity", + Parameters: map[string]interface{}{ + "quality": "${Analyze.output.quality}", + }, + Timeout: "5m", + End: true, + }, + { + Name: "HandleTimeout", + Type: StateTypeFail, + Error: "AnalysisTimeout", + Cause: "Code analysis timed out", + }, + }, + } + + // Marshal + data, err := json.Marshal(spec) + if err != nil { + t.Fatalf("Failed to marshal: %v", err) + } + + // Unmarshal + var spec2 WorkflowSpec + err = json.Unmarshal(data, &spec2) + if err != nil { + t.Fatalf("Failed to unmarshal: %v", err) + } + + // Verify + if spec2.Name != "code-review" { + t.Errorf("Name mismatch") + } + if len(spec2.States) != 4 { + t.Errorf("Expected 4 states, got %d", len(spec2.States)) + } + + // Check first state + if spec2.States[0].Resource != "CloneRepoActivity" { + t.Errorf("First state resource mismatch") + } + + // Check error catching + if len(spec2.States[1].Catch) == 0 { + t.Error("Analyze state should have catch clauses") + } +} diff --git a/k8s/git-commit.yaml b/k8s/git-commit.yaml index a4a4b51..136fd2f 100644 --- a/k8s/git-commit.yaml +++ b/k8s/git-commit.yaml @@ -9,6 +9,6 @@ metadata: app.kubernetes.io/name: poimen app.kubernetes.io/component: orchestrator data: - GIT_COMMIT: "74a8f2e8" # Updated automatically by CI/CD + GIT_COMMIT: "c5994df8" # Updated automatically by CI/CD GIT_BRANCH: "main" DEPLOYMENT_DATE: "2026-08-31" diff --git a/k8s/worker-deployment.yaml b/k8s/worker-deployment.yaml index 73df1f5..41cd055 100644 --- a/k8s/worker-deployment.yaml +++ b/k8s/worker-deployment.yaml @@ -13,7 +13,7 @@ spec: labels: app: poimen-worker annotations: - git-commit: "74a8f2e8" # βœ… Updated on each push, triggers rolling restart + git-commit: "c5994df8" # βœ… Updated on each push, triggers rolling restart deployment-date: "2026-08-31" spec: containers: