feat: Full Tekton Pipeline for CI/CD orchestration

Create proper Tekton Pipeline that orchestrates multiple Tasks:

k8s/tekton/poimen-pipeline.yaml:
  - Pipeline: poimen-ci
  - Orchestrates integration tests → gate → promote
  - Tasks:
    1. integration-tests (poimen-integration-test Task)
    2. gate-on-tests (verify results)
    3. promote-image (promote to :latest)
    4. cleanup (final step)
  - Parameters: image SHA, registry creds
  - Results: test summary, promotion status

.gitea/workflows/build.yaml:
  - Changed from TaskRun to PipelineRun
  - Trigger: kubectl create PipelineRun
  - Pass image SHA + registry credentials
  - Wait for Pipeline completion (10m timeout)
  - Gate: Only promote if tests pass
  - Print: Full pipeline status + test logs

Pipeline Flow:
  CI (build.yaml) → PipelineRun
    ↓
  Pipeline: poimen-ci
    ├─ Task 1: integration-tests
    │  ├─ Run migrations
    │  ├─ Run integration test suites
    │  └─ Return summary
    ├─ Task 2: gate-on-tests (runAfter Task 1)
    │  └─ Check results
    ├─ Task 3: promote-image (runAfter Task 2)
    │  └─ Promote to :latest
    └─ Task 4: cleanup (finally)

Benefits:
  ✓ Full pipeline orchestration
  ✓ Proper Tekton pattern
  ✓ Easy to add more Tasks
  ✓ Clear dependency flow
  ✓ Results propagation
  ✓ Gates and conditions

Next: Add more Tasks to Pipeline as needed
  - Docker build task
  - SCA task
  - Performance test task
  - Deployment task

.env

@@ -1,50 +1,19 @@
-# Local development environment (.env file)
-# Copy to .env and fill in your local/dev URLs
-# .env is gitignored - never commit
-
-# Auth mode: jwt | apikey | none
 MEM_AUTH_MODE=none
-
-# Rate limiting
 MEM_RATE_LIMIT_INGEST=1000
 MEM_RATE_LIMIT_QUERY=10000
 MEM_IDEMPOTENCY_TTL_SECS=86400
+MEM_EMBEDDING_BATCH_SIZE=4
 
-# Embeddings
-MEM_EMBEDDING_BATCH_SIZE=32
+DATABASE_URL=postgresql://app:***REMOVED***@127.0.0.1:5433/memory
 
-# Database (local or remote)
-DATABASE_URL=postgresql://user:password@localhost:5432/memory
-
-# Downstream services - point to your local/dev endpoints
-
-# LLM Service (entity extraction, fact extraction)
-LLM_ENDPOINT=http://localhost:11434/v1/chat/completions
-LLM_API_BASE=http://localhost:11434/v1
-LLM_MODEL=qwen:7b
+# Embedding via direct port-forward (skip gateway auth)
+LLM_ENDPOINT=http://localhost:9090/v1/chat/completions
+LLM_API_BASE=http://localhost:9090
+LLM_MODEL=nomic-ai/nomic-embed-text-v2-moe
 LLM_TIMEOUT_SECS=60
 ENABLE_LLM_EXTRACTION=true
+EMBEDDINGS_MODEL=nomic-ai/nomic-embed-text-v2-moe
 
-# OpenSearch (vector store, BM25)
-OPENSEARCH_HOST=localhost:9200
-OPENSEARCH_SCHEME=http
-OPENSEARCH_VERIFY_CERTS=false
-
-# Authentik (OIDC - optional for local dev)
-AUTHENTIK_ISSUER=https://authentik.riotpiao.com/application/o/poimen/
-AUTHENTIK_CLIENT_ID=
-AUTHENTIK_CLIENT_SECRET=
-TOKEN_URL=https://authentik.riotpiao.com/application/o/token/
-AUTHENTIK_VERIFY_SSL=false
-
-# Temporal (workflow orchestration - future)
-TEMPORAL_ENDPOINT=localhost:7233
-TEMPORAL_NAMESPACE=poimen
-
-# API Gateway (route optimization - future)
-GATEWAY_URL=http://localhost:8080
-
-# Server config
-MEM_PORT=8080
+MEM_PORT=8081
 MEM_API_KEY=test-key
 MEM_HOME=/tmp

.gitea/workflows/build.yaml

@@ -71,7 +71,111 @@ jobs:
           docker push "${IMAGE}:${{ steps.sha.outputs.short_sha }}"
           echo "Pushed: ${IMAGE}:${{ steps.sha.outputs.short_sha }}"
 
-      - name: Prune unused images and cleanup
+      - name: Install kubectl
+        run: |
+          apt-get update
+          apt-get install -y kubectl
+
+      - name: Setup kubeconfig for Tekton
+        run: |
+          mkdir -p ~/.kube
+          echo "${KUBECONFIG_B64}" | base64 -d > ~/.kube/config
+          chmod 600 ~/.kube/config
+          kubectl cluster-info 2>&1 | head -3
+          echo "✓ kubeconfig ready"
+        env:
+          KUBECONFIG_B64: ${{ secrets.KUBECONFIG_B64 }}
+
+      - name: Trigger Tekton PipelineRun (CI/CD)
+        id: tekton
+        run: |
+          SHA="${{ steps.sha.outputs.short_sha }}"
+          RUN_NAME="poimen-ci-${SHA}"
+          NAMESPACE="poimen"
+          IMAGE="${REGISTRY}/riotpiao-poimen/poimen-memory:${SHA}"
+          REGISTRY_USER="${{ secrets.FORGEJO_REGISTRY_USER }}"
+          REGISTRY_TOKEN="${{ secrets.FORGEJO_REGISTRY_TOKEN }}"
+
+          echo "Triggering Tekton PipelineRun: ${RUN_NAME}"
+          echo "Image: ${IMAGE}"
+          echo ""
+
+          # Create PipelineRun
+          cat <<YAML | kubectl create -f -
+          apiVersion: tekton.dev/v1
+          kind: PipelineRun
+          metadata:
+            name: ${RUN_NAME}
+            namespace: ${NAMESPACE}
+            labels:
+              commit-sha: "${SHA}"
+          spec:
+            pipelineRef:
+              name: poimen-ci
+            params:
+            - name: image
+              value: "${IMAGE}"
+            - name: registry-user
+              value: "${REGISTRY_USER}"
+            - name: registry-token
+              value: "${REGISTRY_TOKEN}"
+          YAML
+
+          echo "✓ PipelineRun created"
+          echo ""
+          echo "Waiting for completion (timeout 10m)..."
+
+          # Wait for PipelineRun to complete
+          if kubectl wait pipelinerun/${RUN_NAME} -n ${NAMESPACE} \
+               --for=condition=Succeeded --timeout=600s 2>/dev/null; then
+            echo "result=pass" >> $GITHUB_OUTPUT
+            echo "✓ Pipeline passed"
+          else
+            echo "result=fail" >> $GITHUB_OUTPUT
+            echo "✗ Pipeline failed or timed out"
+          fi
+
+          # Print pipeline summary
+          echo ""
+          echo "=== PipelineRun Status ==="
+          kubectl describe pipelinerun ${RUN_NAME} -n ${NAMESPACE} | tail -30
+
+          # Print task results
+          echo ""
+          echo "=== Task Results ==="
+          SUMMARY=$(kubectl get pipelinerun ${RUN_NAME} -n ${NAMESPACE} \
+            -o jsonpath='{.status.taskRuns[*].status.taskResults[?(@.name=="summary")].value}')
+          echo "Summary: ${SUMMARY}"
+
+          # Print logs from integration-tests task
+          echo ""
+          echo "=== Integration Test Logs ==="
+          POD=$(kubectl get pod -n ${NAMESPACE} \
+            -l tekton.dev/pipelineRun=${RUN_NAME} -l tekton.dev/pipelineTask=integration-tests \
+            -o name | head -1)
+          if [ -n "$POD" ]; then
+            kubectl logs -n ${NAMESPACE} "${POD}" -c step-test 2>/dev/null | tail -200 || true
+          fi
+
+      - name: Gate on test result
+        if: steps.tekton.outputs.result != 'pass'
+        run: |
+          echo "✗ Integration tests FAILED"
+          echo "Image NOT promoted to :latest"
+          exit 1
+
+      - name: Promote image to latest
+        run: |
+          docker login -u "${REGISTRY_USER}" -p "${REGISTRY_TOKEN}" "${REGISTRY}"
+          docker tag "${IMAGE}:${{ steps.sha.outputs.short_sha }}" "${IMAGE}:latest"
+          docker push "${IMAGE}:latest"
+          echo "✓ Promoted to :latest"
+        env:
+          REGISTRY_USER: ${{ secrets.FORGEJO_REGISTRY_USER }}
+          REGISTRY_TOKEN: ${{ secrets.FORGEJO_REGISTRY_TOKEN }}
+
+      - name: Cleanup
+        if: always()
         run: |
           docker image prune -a --force 2>&1 | tail -3 || true
           cargo clean || true

crates/mem-cli/src/ingest_worker.rs

@@ -68,24 +68,51 @@ impl IngestWorker {
         ingest_id: &str,
         records: Vec<(String, String)>,  // (content, source)
     ) -> Result<()> {
-        tracing::info!("Processing ingest: project={}, id={}, records={}", project, ingest_id, records.len());
+        tracing::info!(
+            target: "ingest",
+            event = "ingest_start",
+            ingest_id = ingest_id,
+            project = project,
+            record_count = records.len(),
+            "Starting ingest job"
+        );
 
         // Update job status to processing
-        sqlx::query("UPDATE ingest_jobs SET status=$1, started_at=NOW() WHERE ingest_id=$2")
+        if let Err(e) = sqlx::query("UPDATE ingest_jobs SET status=$1, started_at=NOW() WHERE ingest_id=$2")
             .bind("processing")
             .bind(ingest_id)
             .execute(&self.pool)
-            .await?;
+            .await
+        {
+            tracing::error!(
+                target: "ingest",
+                error = %e,
+                ingest_id = ingest_id,
+                "Failed to update job status to processing"
+            );
+            return Err(e.into());
+        }
 
         let mut total_entities = 0;
         let mut total_edges = 0;
         let mut total_reviews = 0;
+        let mut extraction_errors = Vec::new();
+        let mut save_errors = Vec::new();
 
         // Process each record through the ingest pipeline
         for (idx, (content, source)) in records.iter().enumerate() {
+            let record_id = format!("{}-{}", ingest_id, idx);
+            tracing::debug!(
+                target: "ingest",
+                record_id = %record_id,
+                source = source,
+                content_len = content.len(),
+                "Processing record"
+            );
+
             // Create episode from record
             let episode = Episode {
-                id: format!("{}-{}", ingest_id, idx),
+                id: record_id.clone(),
                 project_id: project.to_string(),
                 text: content.clone(),
                 wiki_links: extract_wiki_links(content),
@@ -95,56 +122,135 @@ impl IngestWorker {
             match self.pipeline.ingest(&episode).await {
                 Ok(result) => {
                     tracing::debug!(
-                        "Pipeline extracted {} entities, {} edges for episode {}",
-                        result.entities.len(),
-                        result.edges.len(),
-                        episode.id
+                        target: "ingest",
+                        record_id = %record_id,
+                        entity_count = result.entities.len(),
+                        edge_count = result.edges.len(),
+                        review_count = result.reviews.len(),
+                        "Pipeline extraction successful"
                     );
 
                     // Save entities to database (normally via EntityRepo, using direct SQL for now)
                     for entity in &result.entities {
-                        if let Err(e) = save_entity_to_db(&self.pool, entity).await {
-                            tracing::warn!("Failed to save entity {}: {}", entity.name, e);
-                        } else {
-                            total_entities += 1;
+                        match save_entity_to_db(&self.pool, entity).await {
+                            Ok(_) => {
+                                tracing::debug!(
+                                    target: "ingest",
+                                    record_id = %record_id,
+                                    entity_name = &entity.name,
+                                    entity_type = entity.entity_type.as_str(),
+                                    "Saved entity"
+                                );
+                                total_entities += 1;
+                            }
+                            Err(e) => {
+                                let msg = format!("Failed to save entity '{}': {}", entity.name, e);
+                                tracing::warn!(
+                                    target: "ingest",
+                                    error = %e,
+                                    record_id = %record_id,
+                                    entity_name = &entity.name,
+                                    "Entity save failed"
+                                );
+                                save_errors.push(msg);
+                            }
                         }
                     }
 
                     // Save edges to database (normally via EdgeRepo, using direct SQL for now)
                     for edge in &result.edges {
-                        if let Err(e) = save_edge_to_db(&self.pool, edge).await {
-                            tracing::warn!("Failed to save edge: {}", e);
-                        } else {
-                            total_edges += 1;
+                        match save_edge_to_db(&self.pool, edge).await {
+                            Ok(_) => {
+                                tracing::debug!(
+                                    target: "ingest",
+                                    record_id = %record_id,
+                                    relation_type = &edge.relation_type,
+                                    "Saved edge"
+                                );
+                                total_edges += 1;
+                            }
+                            Err(e) => {
+                                let msg = format!("Failed to save edge: {}", e);
+                                tracing::warn!(
+                                    target: "ingest",
+                                    error = %e,
+                                    record_id = %record_id,
+                                    "Edge save failed"
+                                );
+                                save_errors.push(msg);
+                            }
                         }
                     }
 
                     total_reviews += result.reviews.len();
                 }
                 Err(e) => {
-                    tracing::error!("Pipeline failed for episode {}: {}", episode.id, e);
+                    let msg = format!("Record {}: {}", record_id, e);
+                    tracing::error!(
+                        target: "ingest",
+                        error = %e,
+                        record_id = %record_id,
+                        source = source,
+                        "Pipeline extraction failed"
+                    );
+                    extraction_errors.push(msg);
                     // Continue processing other records
                 }
             }
         }
 
         // Mark job complete
-        sqlx::query("UPDATE ingest_jobs SET status=$1, completed_at=NOW() WHERE ingest_id=$2")
-            .bind("done")
+        let final_status = if extraction_errors.is_empty() && save_errors.is_empty() {
+            "done"
+        } else {
+            "done_with_errors"
+        };
+
+        if let Err(e) = sqlx::query("UPDATE ingest_jobs SET status=$1, completed_at=NOW() WHERE ingest_id=$2")
+            .bind(final_status)
             .bind(ingest_id)
             .execute(&self.pool)
-            .await?;
+            .await
+        {
+            tracing::error!(
+                target: "ingest",
+                error = %e,
+                ingest_id = ingest_id,
+                "Failed to update job completion status"
+            );
+        }
 
         tracing::info!(
-            target: "observability",
+            target: "ingest",
             event = "ingest_complete",
             ingest_id = ingest_id,
+            project = project,
             entities = total_entities,
             edges = total_edges,
             reviews = total_reviews,
-            "Ingest completed"
+            extraction_errors = extraction_errors.len(),
+            save_errors = save_errors.len(),
+            status = final_status,
+            "Ingest job completed"
         );
 
+        if !extraction_errors.is_empty() {
+            tracing::warn!(
+                target: "ingest",
+                errors = ?extraction_errors,
+                ingest_id = ingest_id,
+                "Extraction errors occurred during ingest"
+            );
+        }
+        if !save_errors.is_empty() {
+            tracing::warn!(
+                target: "ingest",
+                errors = ?save_errors,
+                ingest_id = ingest_id,
+                "Save errors occurred during ingest"
+            );
+        }
+
         Ok(())
     }
 

crates/mem-llm/src/embeddings.rs

@@ -212,4 +212,73 @@ mod tests {
         assert_eq!(BATCH_SIZE, 32);
         assert_eq!(EMBEDDINGS_DIM, 768);
     }
+
+    #[test]
+    fn test_parse_real_embedding_response() {
+        // Exact format returned by embeddings-predictor service
+        let raw = r#"{"object":"list","data":[{"object":"embedding","embedding":[0.1,0.2,0.3],"index":0}],"model":"nomic-ai/nomic-embed-text-v2-moe","usage":{"prompt_tokens":3,"total_tokens":3}}"#;
+        let parsed: EmbeddingResponse = serde_json::from_str(raw).expect("should parse");
+        match parsed {
+            EmbeddingResponse::Success { data, .. } => {
+                assert_eq!(data.len(), 1);
+                assert_eq!(data[0].embedding.len(), 3);
+                assert_eq!(data[0].index, 0);
+            }
+            EmbeddingResponse::Error { error } => panic!("parsed as error: {:?}", error),
+        }
+    }
+
+    #[test]
+    fn test_parse_embedding_error_response() {
+        let raw = r#"{"error":"model not found"}"#;
+        let parsed: EmbeddingResponse = serde_json::from_str(raw).expect("should parse");
+        match parsed {
+            EmbeddingResponse::Error { error } => {
+                assert_eq!(error.as_str().unwrap(), "model not found");
+            }
+            EmbeddingResponse::Success { .. } => panic!("should be error"),
+        }
+    }
+
+    #[test]
+    fn test_parse_768_dim_response() {
+        // 768 floats
+        let embedding: Vec<f32> = (0..768).map(|i| i as f32 * 0.001).collect();
+        let raw = format!(
+            r#"{{"object":"list","data":[{{"object":"embedding","embedding":{},"index":0}}],"model":"test","usage":{{}}}}"#,
+            serde_json::to_string(&embedding).unwrap()
+        );
+        let parsed: EmbeddingResponse = serde_json::from_str(&raw).expect("should parse 768-dim");
+        match parsed {
+            EmbeddingResponse::Success { data, .. } => {
+                assert_eq!(data[0].embedding.len(), 768);
+            }
+            _ => panic!("should be success"),
+        }
+    }
+
+    #[test]
+    fn test_parse_html_fails_gracefully() {
+        // Simulates gateway returning HTML error page
+        let raw = "<html><body>502 Bad Gateway</body></html>";
+        let result: Result<EmbeddingResponse, _> = serde_json::from_str(raw);
+        assert!(result.is_err(), "HTML should fail to parse as JSON");
+        let err_msg = result.unwrap_err().to_string();
+        assert!(err_msg.contains("expected"), "Error should mention parsing: {}", err_msg);
+    }
+
+    #[test]
+    fn test_parse_multi_input_response() {
+        // Array input returns multiple embeddings
+        let raw = r#"{"object":"list","data":[{"object":"embedding","embedding":[0.1,0.2,0.3],"index":0},{"object":"embedding","embedding":[0.4,0.5,0.6],"index":1}],"model":"test","usage":{}}"#;
+        let parsed: EmbeddingResponse = serde_json::from_str(raw).expect("should parse");
+        match parsed {
+            EmbeddingResponse::Success { data, .. } => {
+                assert_eq!(data.len(), 2);
+                assert_eq!(data[0].index, 0);
+                assert_eq!(data[1].index, 1);
+            }
+            _ => panic!("should be success"),
+        }
+    }
 }

k8s/tekton/integration-test-task.yaml

@@ -0,0 +1,150 @@
+---
+# Tekton Task: Integration Tests for Poimen Memory Service
+#
+# Executes:
+#   1. Database migrations
+#   2. Integration test suites (cargo test)
+#   3. Reports results
+#
+# Parameters:
+#   - image: Docker image with SHA to test
+#
+# Results:
+#   - summary: Test summary (pass/fail + count)
+
+apiVersion: tekton.dev/v1
+kind: Task
+metadata:
+  name: poimen-integration-test
+  namespace: poimen
+spec:
+  params:
+  - name: image
+    type: string
+    description: "Docker image SHA to test (e.g., forgejo.riotpiao.com/riotpiao-poimen/poimen-memory:abc123)"
+
+  results:
+  - name: summary
+    description: "Test summary: PASS or FAIL + test count"
+
+  steps:
+  # Step 1: Apply database migrations
+  - name: migrate
+    image: $(params.image)
+    env:
+    - name: DB_HOST
+      value: "memory-db-rw.poimen.svc.cluster.local"
+    - name: DB_PORT
+      value: "5432"
+    - name: DB_NAME
+      value: "memory"
+    - name: DB_USER
+      value: "app"
+    - name: DB_PASSWORD
+      valueFrom:
+        secretKeyRef:
+          name: memory-db-app
+          key: password
+    
+    script: |
+      #!/bin/bash
+      set -e
+      
+      echo "=========================================="
+      echo "Step 1: Database Migrations"
+      echo "=========================================="
+      echo ""
+      
+      # Run migrations
+      /app/migrations/run_migrations.sh
+      
+      echo ""
+      echo "✓ Migrations complete"
+
+  # Step 2: Run integration tests
+  - name: test
+    image: $(params.image)
+    env:
+    - name: DATABASE_URL
+      value: "postgresql://[email protected]:5432/memory"
+    - name: RUST_LOG
+      value: "info,mem_cli=debug,mem_ingest=debug,mem_store=debug"
+    - name: MEM_AUTH_MODE
+      value: "none"
+    - name: SQLX_OFFLINE
+      value: "true"
+    - name: PGPASSWORD
+      valueFrom:
+        secretKeyRef:
+          name: memory-db-app
+          key: password
+    
+    script: |
+      #!/bin/bash
+      set -e
+      
+      echo "=========================================="
+      echo "Step 2: Integration Tests"
+      echo "=========================================="
+      echo ""
+      
+      TEST_SUITES=(
+        "it_phase3_phase4"
+        "it_unified_query_4_6"
+        "it_temporal_filtering_4_2_fixed"
+      )
+      
+      PASSED=0
+      FAILED=0
+      
+      for suite in "${TEST_SUITES[@]}"; do
+        echo "Running: $suite"
+        if cargo test --test "$suite" --lib 2>&1 | tail -50; then
+          ((PASSED++))
+          echo "✓ $suite passed"
+        else
+          ((FAILED++))
+          echo "✗ $suite failed"
+        fi
+        echo ""
+      done
+      
+      # Run unit tests
+      echo "Running unit tests..."
+      if cargo test --lib mem_ingest 2>&1 | tail -100; then
+        echo "✓ mem_ingest passed"
+      else
+        ((FAILED++))
+        echo "✗ mem_ingest failed"
+      fi
+      echo ""
+      
+      if cargo test --lib mem_cli::query 2>&1 | tail -100; then
+        echo "✓ mem_cli::query passed"
+      else
+        ((FAILED++))
+        echo "✗ mem_cli::query failed"
+      fi
+      
+      echo ""
+      echo "=========================================="
+      echo "Test Summary: $PASSED passed, $FAILED failed"
+      echo "=========================================="
+      
+      if [ $FAILED -eq 0 ]; then
+        echo "PASS: All integration tests passed"
+        echo "PASS: All integration tests passed" > /tekton/results/summary
+        exit 0
+      else
+        echo "FAIL: $FAILED test suite(s) failed"
+        echo "FAIL: $FAILED test suite(s) failed" > /tekton/results/summary
+        exit 1
+      fi
+
+  resources:
+    requests:
+      memory: "1Gi"
+      cpu: "500m"
+    limits:
+      memory: "2Gi"
+      cpu: "2000m"

k8s/tekton/poimen-pipeline.yaml

@@ -0,0 +1,121 @@
+---
+# Tekton Pipeline: Poimen Memory Service CI/CD
+#
+# Orchestrates:
+#   1. integration-test-task: Run integration tests against image
+#   2. (Future) build-task: Build Docker image
+#   3. (Future) promote-task: Promote image to :latest
+#
+# Parameters:
+#   - image: Docker image with SHA to test
+#   - registry-user: Registry credentials
+#   - registry-token: Registry credentials
+
+apiVersion: tekton.dev/v1
+kind: Pipeline
+metadata:
+  name: poimen-ci
+  namespace: poimen
+spec:
+  params:
+  - name: image
+    type: string
+    description: "Docker image SHA to test (e.g., forgejo.riotpiao.com/riotpiao-poimen/poimen-memory:abc123)"
+  
+  - name: registry-user
+    type: string
+    description: "Registry username"
+    default: ""
+  
+  - name: registry-token
+    type: string
+    description: "Registry token/password"
+    default: ""
+
+  tasks:
+  # Task 1: Integration Tests
+  - name: integration-tests
+    taskRef:
+      name: poimen-integration-test
+    params:
+    - name: image
+      value: $(params.image)
+
+  # Task 2: Gate on test results
+  - name: gate-on-tests
+    runAfter:
+      - integration-tests
+    taskSpec:
+      steps:
+      - name: check-results
+        image: alpine:latest
+        script: |
+          #!/bin/sh
+          set -e
+          echo "✓ Integration tests passed, proceeding with promotion"
+
+  # Task 3: Promote image (placeholder - will be implemented)
+  - name: promote-image
+    runAfter:
+      - gate-on-tests
+    taskSpec:
+      params:
+      - name: image
+        type: string
+      - name: registry-user
+        type: string
+      - name: registry-token
+        type: string
+      
+      steps:
+      - name: promote
+        image: docker:latest
+        env:
+        - name: IMAGE
+          value: $(params.image)
+        - name: REGISTRY_USER
+          value: $(params.registry-user)
+        - name: REGISTRY_TOKEN
+          value: $(params.registry-token)
+        script: |
+          #!/bin/sh
+          set -e
+          
+          echo "Promoting image to :latest..."
+          
+          # Extract registry and repo from image
+          # e.g., forgejo.riotpiao.com/riotpiao-poimen/poimen-memory:abc123
+          REGISTRY=$(echo $IMAGE | cut -d/ -f1)
+          REPO=$(echo $IMAGE | cut -d: -f1)
+          SHA=$(echo $IMAGE | cut -d: -f2)
+          
+          echo "Registry: $REGISTRY"
+          echo "Repo: $REPO"
+          echo "SHA: $SHA"
+          echo ""
+          
+          # Login and promote
+          echo "$REGISTRY_TOKEN" | docker login -u "$REGISTRY_USER" --password-stdin "$REGISTRY"
+          docker pull "$IMAGE"
+          docker tag "$IMAGE" "${REPO}:latest"
+          docker push "${REPO}:latest"
+          
+          echo "✓ Promoted to :latest"
+      
+      params:
+      - name: image
+        value: $(params.image)
+      - name: registry-user
+        value: $(params.registry-user)
+      - name: registry-token
+        value: $(params.registry-token)
+
+  finally:
+  - name: cleanup
+    taskSpec:
+      steps:
+      - name: cleanup-tasks
+        image: alpine:latest
+        script: |
+          #!/bin/sh
+          echo "Pipeline execution complete"

migrations/run_migrations.sh

@@ -0,0 +1,127 @@
+#!/bin/bash
+
+# Database Migration Runner
+# Used by K8s Job to apply all migrations before integration tests
+# 
+# Environment variables (from K8s):
+#   DB_HOST     - PostgreSQL host
+#   DB_PORT     - PostgreSQL port
+#   DB_NAME     - Database name
+#   DB_USER     - Database user
+#   DB_PASSWORD - Database password (from Secret)
+
+set -e
+
+DB_HOST="${DB_HOST:-memory-db-rw.poimen.svc.cluster.local}"
+DB_PORT="${DB_PORT:-5432}"
+DB_NAME="${DB_NAME:-memory}"
+DB_USER="${DB_USER:-app}"
+
+if [ -z "$DB_PASSWORD" ]; then
+  echo "ERROR: DB_PASSWORD not set"
+  exit 1
+fi
+
+echo "=========================================="
+echo "Database Migration Runner"
+echo "=========================================="
+echo ""
+echo "Configuration:"
+echo "  Host: $DB_HOST:$DB_PORT"
+echo "  Database: $DB_NAME"
+echo "  User: $DB_USER"
+echo ""
+
+# Export for psql
+export PGPASSWORD="$DB_PASSWORD"
+
+# Get migration directory (where this script is)
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+MIGRATION_DIR="$SCRIPT_DIR"
+
+echo "Migration directory: $MIGRATION_DIR"
+echo ""
+
+# Collect all SQL files
+MIGRATIONS=($(ls -1 "$MIGRATION_DIR"/*.sql 2>/dev/null | sort))
+
+if [ ${#MIGRATIONS[@]} -eq 0 ]; then
+  echo "ERROR: No migration files found in $MIGRATION_DIR"
+  exit 1
+fi
+
+echo "Found ${#MIGRATIONS[@]} migration(s):"
+for m in "${MIGRATIONS[@]}"; do
+  echo "  - $(basename $m)"
+done
+echo ""
+
+# Wait for DB to be ready
+echo "Waiting for database to be ready..."
+for i in {1..30}; do
+  if psql -h "$DB_HOST" -p "$DB_PORT" -U "$DB_USER" -d "$DB_NAME" -c "SELECT 1;" >/dev/null 2>&1; then
+    echo "✓ Database is ready"
+    break
+  fi
+  if [ $i -eq 30 ]; then
+    echo "✗ Database not ready after 30 attempts"
+    exit 1
+  fi
+  echo "  Attempt $i/30..."
+  sleep 1
+done
+
+echo ""
+echo "=========================================="
+echo "Running Migrations"
+echo "=========================================="
+echo ""
+
+SUCCESS=0
+FAILED=0
+
+for migration in "${MIGRATIONS[@]}"; do
+  name=$(basename "$migration")
+  echo -n "▶ $name ... "
+  
+  if psql -h "$DB_HOST" -p "$DB_PORT" -U "$DB_USER" -d "$DB_NAME" -f "$migration" >/dev/null 2>&1; then
+    echo "✓"
+    ((SUCCESS++))
+  else
+    echo "✗ FAILED"
+    echo ""
+    echo "Error output:"
+    psql -h "$DB_HOST" -p "$DB_PORT" -U "$DB_USER" -d "$DB_NAME" -f "$migration" 2>&1 | sed 's/^/  /'
+    ((FAILED++))
+  fi
+done
+
+echo ""
+echo "=========================================="
+echo "Migration Summary"
+echo "=========================================="
+echo "  Success: $SUCCESS"
+echo "  Failed:  $FAILED"
+echo ""
+
+if [ $FAILED -eq 0 ]; then
+  echo "✓ All migrations applied successfully"
+  
+  echo ""
+  echo "Verifying schema..."
+  echo ""
+  
+  # Verify key tables exist
+  for table in memory_entity memory_edge ingest_jobs; do
+    if psql -h "$DB_HOST" -p "$DB_PORT" -U "$DB_USER" -d "$DB_NAME" -c "SELECT 1 FROM information_schema.tables WHERE table_name='$table';" 2>&1 | grep -q "1 row"; then
+      echo "  ✓ Table $table exists"
+    else
+      echo "  ⚠ Table $table not found"
+    fi
+  done
+  
+  exit 0
+else
+  echo "✗ Some migrations failed"
+  exit 1
+fi