fix: install kubectl from upstream release instead of apt

kubectl not in default Debian repos, download from Google release
fix: resolve mem-cli build errors
2026-09-15 02:49:03 +09:00 · 2026-09-15 02:24:53 +09:00 · 2026-09-15 00:44:05 +09:00 · 2026-09-15 00:43:31 +09:00 · 2026-09-15 00:36:05 +09:00 · 2026-09-15 00:34:40 +09:00
51 changed files with 2614 additions and 908 deletions
@@ -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
+DATABASE_URL=postgresql://app:***REMOVED***@127.0.0.1:5433/memory
 MEM_EMBEDDING_BATCH_SIZE=32
-# Database (local or remote)
+# Embedding via direct port-forward (skip gateway auth)
-DATABASE_URL=postgresql://user:password@localhost:5432/memory
+LLM_ENDPOINT=http://localhost:9090/v1/chat/completions
-
+LLM_API_BASE=http://localhost:9090
-# Downstream services - point to your local/dev endpoints
+LLM_MODEL=nomic-ai/nomic-embed-text-v2-moe
 # 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
 LLM_TIMEOUT_SECS=60
 ENABLE_LLM_EXTRACTION=true
 EMBEDDINGS_MODEL=nomic-ai/nomic-embed-text-v2-moe
-# OpenSearch (vector store, BM25)
+MEM_PORT=8081
 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_API_KEY=test-key
 MEM_HOME=/tmp
@@ -71,7 +71,114 @@ 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 curl
          curl -LO "https://dl.k8s.io/release/$(curl -L -s https://dl.k8s.io/release/stable.txt)/bin/linux/amd64/kubectl"
          chmod +x kubectl
          mv kubectl /usr/local/bin/
      - 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
@@ -1,85 +0,0 @@
 name: DB Migration
 on:
  push:
    branches: [main]
    paths:
      - 'crates/mem-store/migrations/**'
  workflow_dispatch:
 env:
  DB_HOST: memory-db-rw.poimen.svc.cluster.local
  DB_PORT: "5432"
  DB_NAME: memory
 jobs:
  migrate:
    name: Run Migrations
    runs-on: rust
    steps:
      - name: Install psql
        run: apt-get update && apt-get install -y postgresql-client
      - name: Checkout code
        uses: actions/checkout@v4
      - name: Fetch previous migrations state
        run: |
          git fetch origin main --depth=2
          # List changed migration files
          CHANGED=$(git diff --name-only HEAD~1 HEAD -- crates/mem-store/migrations/ || echo "")
          echo "Changed migrations: $CHANGED"
          echo "CHANGED_MIGRATIONS=$CHANGED" >> $GITHUB_ENV
      - name: Run changed migrations and verify schema
        if: env.CHANGED_MIGRATIONS != ''
        run: |
          export PGPASSWORD="${DB_PASSWORD}"
          echo "=== Running changed migrations ==="
          for f in $CHANGED_MIGRATIONS; do
            if [ -f "$f" ]; then
              echo "--- Applying: $f ---"
              psql -h "$DB_HOST" -p "$DB_PORT" -U "$DB_USER" -d "$DB_NAME" -f "$f" 2>&1
              if [ $? -ne 0 ]; then
                echo "ERROR: Migration $f failed!"
                exit 1
              fi
              echo "--- OK: $f ---"
            fi
          done
          echo "=== Verify schema ==="
          psql -h "$DB_HOST" -p "$DB_PORT" -U "$DB_USER" -d "$DB_NAME" -c "\dt memory*"
        env:
          DB_USER: ${{ secrets.DB_USER }}
          DB_PASSWORD: ${{ secrets.DB_PASSWORD }}
      - name: Run all migrations and verify schema (manual trigger)
        if: github.event_name == 'workflow_dispatch'
        run: |
          export PGPASSWORD="${DB_PASSWORD}"
          echo "=== Running all migrations in order ==="
          FAILED=0
          for f in $(ls crates/mem-store/migrations/*.sql | sort); do
            echo "--- Applying: $f ---"
            if ! psql -h "$DB_HOST" -p "$DB_PORT" -U "$DB_USER" -d "$DB_NAME" -f "$f" 2>&1; then
              echo "ERROR: Migration $f failed!"
              FAILED=1
            else
              echo "--- OK: $f ---"
            fi
          done
          if [ $FAILED -eq 1 ]; then
            exit 1
          fi
          echo "=== Final schema ==="
          psql -h "$DB_HOST" -p "$DB_PORT" -U "$DB_USER" -d "$DB_NAME" -c "\dt memory*"
          psql -h "$DB_HOST" -p "$DB_PORT" -U "$DB_USER" -d "$DB_NAME" -c "\d memory_entity"
          psql -h "$DB_HOST" -p "$DB_PORT" -U "$DB_USER" -d "$DB_NAME" -c "\d memory_edge"
        env:
          DB_USER: ${{ secrets.DB_USER }}
          DB_PASSWORD: ${{ secrets.DB_PASSWORD }}
@@ -1,263 +0,0 @@
 # CRITICAL FIXES NEEDED - Poimen Memory Service
 ## STATUS: Service Non-Functional ❌
 **Root Issues Blocking Service**:
 1. ✅ HTTP handler deadlock fixed (schema init error handling)
 2. ❌ Server initialization hangs during schema or startup (logs stop after `l2_l1_edges`)
 3. ❌ Ingest pipeline NOT implemented (just raw vector storage, no entities/edges)
 4. ❌ Temporal schema missing (no t_valid, t_invalid, version tracking)
 5. ❌ GRM gate not integrated (no memorability scores, confidence)
 6. ❌ Query doesn't use knowledge graph (just vector search)
 7. ❌ Compaction disabled
 8. ❌ Verification gates missing
 ---
 ## STEP 1: Fix Server Startup Hang ⚠️
 **Current Issue**: Server hangs during initialization after schema creation.
 **Suspected causes**:
 - OptimizerServiceBuilder.build() getting stuck
 - AccessGuard creation blocking
 - Background task spawning deadlock
 **Fix**:
 ```rust
 // In http_server.rs:316-325
 // Wrap in timeout or disable non-essentials
 let optimizer_service = match tokio::time::timeout(
    Duration::from_secs(5),
    async { mem_core::optimizer::OptimizerServiceBuilder::new().build() }
 ).await {
    Ok(Ok(service)) => Some(Arc::new(service)),
    _ => {
        tracing::warn!("Optimizer initialization skipped (timeout or error)");
        None
    }
 };
 ```
 **Test**: `./target/release/mem serve --port 9999` should reach "Starting HTTP server" within 10s
 ---
 ## STEP 2: Implement Ingest Pipeline (HIGH PRIORITY)
 **Current Implementation** (`ingest_worker.rs`):
 ```rust
 // Just stores raw chunks + embeddings
 store_chunk_l0(&l0_chunk)
 store_memory_l1(&l1_memory, &embedding)
 ```
 **Expected Implementation**:
 ```rust
 // 1. Extract entities (entity_extractor)
 let entities = entity_extractor.extract(&content).await?;
 // 2. Extract facts + edges (fact_extractor)  
 let facts = fact_extractor.extract(&content, entities).await?;
 // 3. Create temporal edges with GRM gate
 for fact in facts {
  let edge = TemporalEdge {
    source: fact.source_entity,
    target: fact.target_entity,
    relation: fact.relation,
    fact: fact.text,
    t_valid: now(),
    t_invalid: None,
    confidence: grm_gate.score(&fact)?,  // ← GRM gate
    version: 1,
  };
  edge_repo.insert(&edge).await?;
 }
 // 4. Check contradictions + queue for review
 for edge in edges {
  if contradiction_detector.detect(&edge, existing_edges)? {
    review_queue.enqueue(&edge).await?;
  }
 }
 ```
 **Files to modify**:
 - `crates/mem-cli/src/ingest_worker.rs` (core ingest logic)
 - `crates/mem-ingest/src/ingest_pipeline.rs` (entity + fact extraction)
 - `crates/mem-ingest/src/contradiction_detector.rs` (pre-filter + review)
 ---
 ## STEP 3: Update Storage Schema (MEDIUM PRIORITY)
 **Missing fields**:
 ```sql
 ALTER TABLE memories_l1 ADD COLUMN (
  t_valid TIMESTAMP NOT NULL DEFAULT NOW(),
  t_invalid TIMESTAMP,
  confidence FLOAT DEFAULT 0.5,
  version INT DEFAULT 1,
  memorability_score INT,
  contribution_date TIMESTAMP
 );
 ALTER TABLE l1_l0_edges MODIFY TO (
  l1_id UUID,
  l0_id UUID,
  relation_type VARCHAR,
  fact TEXT,
  t_valid TIMESTAMP DEFAULT NOW(),
  t_invalid TIMESTAMP,
  confidence FLOAT,
  contradiction_flag BOOL DEFAULT FALSE,
  review_queue_id UUID,
  version INT DEFAULT 1,
  PRIMARY KEY (l1_id, l0_id, version)
 );
 ```
 **Migration script**: `crates/mem-store/migrations/003_temporal_grm_schema.sql`
 ---
 ## STEP 4: Wire Query Handler to Knowledge Graph (MEDIUM PRIORITY)
 **Current** (`query_handler` in http_server.rs):
 ```rust
 async fn query_handler(...) -> HttpResponse {
    // Just semantic search
    let results = vector_search(query)?;
    HttpResponse::Ok().json(results)
 }
 ```
 **Expected**:
 ```rust
 async fn query_handler(query: QueryRequest) -> HttpResponse {
    // 1. Semantic search on embeddings
    let initial_results = vector_search(&query.text)?;
    // 2. Follow edges (graph traversal)
    let mut expanded = vec![];
    for result in initial_results {
        expanded.push(result);
        // Get related entities via edges
        let related = edge_repo.find_by_source(&result.entity_id).await?;
        expanded.extend(related);
    }
    // 3. Apply temporal filters
    expanded.retain(|e| e.t_valid <= now() && (e.t_invalid.is_none() || e.t_invalid > now()));
    // 4. Sort by confidence + recency
    expanded.sort_by(|a, b| {
        b.confidence.partial_cmp(&a.confidence)
            .then_with(|| b.t_valid.cmp(&a.t_valid))
    });
    // 5. Apply compaction/cache alignment
    for item in &mut expanded {
        item.text = optimizer.compress(item.text)?;
    }
    HttpResponse::Ok().json(MemoryResponse {
        entities: expanded,
        confidence_scores: compute_scores(&expanded),
    })
 }
 ```
 ---
 ## STEP 5: Enable Compaction Endpoint (LOW PRIORITY)
 **Current**: Code exists but never called.
 **Fix**: Add K8s CronJob that calls `POST /memory/compact` daily:
 ```yaml
 apiVersion: batch/v1
 kind: CronJob
 metadata:
  name: memory-compaction
 spec:
  schedule: "0 2 * * *"  # 2 AM UTC
  jobTemplate:
    spec:
      template:
        spec:
          containers:
          - name: compact
            image: bitnami/curl:latest
            command:
            - curl
            - -X POST
            - -H "Authorization: Bearer $ADMIN_TOKEN"
            - http://poimen-memory:8080/memory/compact
          restartPolicy: OnFailure
 ```
 ---
 ## STEP 6: Add Verification Gates (LOW PRIORITY)
 **Missing**: `GET /memory/verify` endpoint that checks M1.8, M2.8, M3.7, M8.9 gates
 ---
 ## IMPLEMENTATION ORDER
 1. **FIX STARTUP** (1 hour) → Get server running
 2. **INGEST PIPELINE** (3 hours) → Wire entity + fact extraction
 3. **TEMPORAL SCHEMA** (1 hour) → Add missing columns
 4. **QUERY HANDLER** (2 hours) → Implement graph traversal
 5. **COMPACTION** (1 hour) → Add CronJob
 6. **GATES** (2 hours) → Quality verification
 **Total**: ~10 hours to full working system
 ---
 ## TEST PLAN
 ```bash
 # 1. Server starts
 curl http://localhost:9999/health
 # Expected: {"status":"ok","uptime_seconds":N}
 # 2. Ingest works
 curl -X POST http://localhost:9999/memory/ingest \
  -H "Content-Type: application/json" \
  -d '{"project":"test","source":"test://1","ingest_id":"i1","records":[{"role":"user","text":"Hello world","timestamp":"2026-01-08T16:00:00Z","source_position":0}]}'
 # Expected: {"ingest_id":"i1","status":"pending",...}
 # 3. Query returns entities with edges
 curl -X POST http://localhost:9999/memory/query \
  -H "Content-Type: application/json" \
  -d '{"project":"test","query":"hello"}'
 # Expected: {"results":[{"type":"entity","name":"...","edges":[...]}]}
 # 4. Temporal filtering works
 curl http://localhost:9999/memory/query?project=test&temporal_floor=2026-01-01
 # 5. Compaction works
 curl -X POST http://localhost:9999/memory/compact
 # Expected: {"phase":"completed","records_deduplicated":N}
 ```
 ---
 ## FILES MODIFIED SO FAR
 ✅ `crates/mem-cli/src/http_server.rs` - Added error handling for schema init
 ---
 ## NEXT SESSION TODO
 - [ ] Fix server startup hang (debug OptimizerService)
 - [ ] Implement ingest_worker to call entity_extractor + fact_extractor
 - [ ] Add temporal columns to schema
 - [ ] Update query_handler to traverse edges
 - [ ] Test end-to-end with sample data
@@ -1,217 +0,0 @@
 # Monitoring Agent: Implementation Tasks
 **Milestone**: `monitoring-agent`
 **Status**: 🔧 Not started
 **Duration**: 4-6 weeks
 **Effort**: ~1,500 LOC
 ---
 ## Phase 1: Temporal Setup (3-5 days)
 ### Task 1.1: Deploy Temporal Server in K8s
 - [ ] StatefulSet configuration (persistence)
 - [ ] PostgreSQL event log backend
 - [ ] ElasticSearch for visibility
 - [ ] K8s manifests in `k8s/temporal/`
 - [ ] Health checks + readiness probes
 - **Effort**: 150 LOC | **Time**: 2 days
 - **Dependencies**: None
 - **Blocks**: Phase 2
 ### Task 1.2: Add Temporal SDK to Rust Project
 - [ ] Add `temporal-rust-sdk` to `Cargo.toml`
 - [ ] Create `crates/mem-temporal/` workspace crate
 - [ ] Worker registration + gRPC connection
 - [ ] Activity executor setup
 - [ ] Workflow executor setup
 - **Effort**: 200 LOC | **Time**: 1 day
 - **Dependencies**: 1.1
 - **Blocks**: Phase 2
 ### Task 1.3: Temporal Configuration + Secrets
 - [ ] Environment variables (TEMPORAL_HOST, TEMPORAL_NAMESPACE)
 - [ ] Worker identity configuration
 - [ ] Task queue setup (synthesis-queue, compaction-queue)
 - **Effort**: 50 LOC | **Time**: 4 hours
 - **Dependencies**: 1.1, 1.2
 - **Blocks**: Phase 2
 ---
 ## Phase 2: Agent Workflows (1-2 weeks)
 ### Task 2.1: Synthesis Workflow Definition
 - [ ] `crates/mem-temporal/src/workflows/synthesis_workflow.rs`
 - [ ] Workflow orchestration logic
 - [ ] Activity composition (health check → synthesis → logging → metrics)
 - [ ] Retry policies (exponential backoff, max 5 retries)
 - [ ] Heartbeat configuration (every 10s)
 - **Effort**: 200 LOC | **Time**: 3 days
 - **Dependencies**: 1.2, 1.3
 - **Blocks**: 2.3, 2.4
 ### Task 2.2: Synthesis Activities (5 activities)
 - [ ] `MonitorMemoryHealth` activity
  - GET /health check
  - Latency measurement
  - Failure detection
 - [ ] `ExecuteSynthesis` activity
  - POST /memory/synthesize call
  - LLM integration
  - Heartbeat emission
 - [ ] `LogSynthesisResult` activity
  - POST /memory/ingest (audit)
  - Temporal audit trail
 - [ ] `UpdateCacheMetrics` activity
  - Metric recording
  - Performance tracking
 - [ ] `CoordinateCompaction` activity
  - Signal to compaction agent
  - Readiness check
 - **Effort**: 250 LOC | **Time**: 4 days
 - **Dependencies**: 2.1
 - **Blocks**: 2.3
 ### Task 2.3: Compaction Workflow Definition
 - [ ] `crates/mem-temporal/src/workflows/compaction_workflow.rs`
 - [ ] 4-stage orchestration (identify → dedup → gc → invalidate)
 - [ ] Failure handling + rollback strategy
 - **Effort**: 150 LOC | **Time**: 2 days
 - **Dependencies**: 1.2, 1.3
 - **Blocks**: 2.4
 ### Task 2.4: Compaction Activities (4 activities)
 - [ ] `IdentifyDuplicates` activity
 - [ ] `DeduplicateEdges` activity
 - [ ] `GarbageCollection` activity
 - [ ] `InvalidateCache` activity
 - **Effort**: 200 LOC | **Time**: 3 days
 - **Dependencies**: 2.3
 - **Blocks**: Integration tests
 ### Task 2.5: Worker + Task Queue Registration
 - [ ] Activity worker setup
 - [ ] Workflow worker setup
 - [ ] Task queue polling
 - [ ] Namespace configuration
 - **Effort**: 100 LOC | **Time**: 1 day
 - **Dependencies**: 2.1-2.4
 - **Blocks**: Phase 3
 ---
 ## Phase 3: Agent Self-Awareness (2-3 weeks)
 ### Task 3.1: AGENT_PROMPT Entity Type
 - [ ] Schema: New entity type in memory_entity
 - [ ] Repository: `synthesis_cache_repo.rs` (get_agent_prompt)
 - [ ] Migration: Add to entity type enum
 - [ ] Activity: Load prompt on agent startup
 - **Effort**: 100 LOC | **Time**: 1 day
 - **Dependencies**: Memory service
 - **Blocks**: 3.2
 ### Task 3.2: AGENT_SKILL Linking
 - [ ] Edge type: agent → skill relationships
 - [ ] Repository methods: link_agent_to_skill, get_agent_skills
 - [ ] Confidence tracking per skill
 - [ ] Success rate calculation
 - **Effort**: 80 LOC | **Time**: 1 day
 - **Dependencies**: 3.1
 - **Blocks**: 3.4
 ### Task 3.3: AGENT_PERFORMANCE Metrics
 - [ ] Entity type: Temporal metrics
 - [ ] Repository: Store + query metrics
 - [ ] Activity: Log performance data post-execution
 - [ ] Time window filtering (last_7_days, last_30_days)
 - **Effort**: 120 LOC | **Time**: 2 days
 - **Dependencies**: 3.1
 - **Blocks**: 3.4
 ### Task 3.4: Agent Decision Tracking + Learning
 - [ ] Edge type: agent_decision_outcome
 - [ ] Decision logging (parameter, value, confidence before)
 - [ ] Outcome recording (result, metric)
 - [ ] Confidence evolution (update after outcome)
 - [ ] Learning loop in agent code
 - **Effort**: 200 LOC | **Time**: 3 days
 - **Dependencies**: 3.1-3.3
 - **Blocks**: 3.5
 ### Task 3.5: Agent Audit Trail Integration
 - [ ] Dual audit: Temporal history + Memory entities
 - [ ] Query interface for reviewers
 - [ ] Temporal CLI integration
 - [ ] Retention policy (365 days)
 - **Effort**: 100 LOC | **Time**: 1 day
 - **Dependencies**: 3.1-3.4
 - **Blocks**: Testing
 ---
 ## Testing & Documentation
 ### Task 4.1: Integration Tests
 - [ ] Workflow execution end-to-end
 - [ ] Activity retry behavior
 - [ ] Heartbeat detection
 - [ ] Failure recovery
 - [ ] State replay on restart
 - **Effort**: 300 LOC | **Time**: 3 days
 - **Dependencies**: Phase 2 complete
 - **Blocks**: Integration
 ### Task 4.2: Monitoring & Observability
 - [ ] Temporal UI setup (temporal.riotpiao.com)
 - [ ] Prometheus metrics export
 - [ ] Alerting rules (workflow timeout, activity failure)
 - [ ] Grafana dashboards
 - **Effort**: 150 LOC | **Time**: 2 days
 - **Dependencies**: Phase 1 complete
 - **Blocks**: Production
 ### Task 4.3: Documentation
 - [ ] Agent architecture diagram
 - [ ] Workflow execution flow
 - [ ] Operational runbook
 - [ ] Troubleshooting guide
 - **Effort**: 50 LOC | **Time**: 1 day
 - **Dependencies**: All phases
 - **Blocks**: Release
 ---
 ## Credentials Status
 ✅ **SOPS Encrypted**: `k8s/app/memory-agent-secrets.enc.yaml`
 - CLIENT_ID: `memory-agent`
 - CLIENT_SECRET: Encrypted
 - TOKEN_URL: `https://authentik.riotpiao.com/application/o/token/`
 - AUTHENTIK_ISSUER: `https://authentik.riotpiao.com/application/o/memory-agent/`
 ✅ **JWT Auth Verified**: `memory-agent` credentials working
 - Test result: Token obtained successfully
 - Expiry: 1 hour (3600s)
 - Scopes: Default (sufficient for LLM operations)
 ---
 ## Timeline
 ```
 Week 1 (Phase 1):        Temporal setup
 Week 2-3 (Phase 2):      Agent workflows
 Week 4-5 (Phase 3):      Self-awareness
 Week 6 (Testing + Docs): Integration + release
 ```
 **Start Date**: TBD
 **Target End Date**: TBD (+4-6 weeks)
@@ -1,191 +0,0 @@
 # Current Status - Poimen Memory Service (2026-01-08)
 ## ✅ COMPLETED THIS SESSION
 ### 1. Removed AccessGuard RBAC (Blocker Issue #1)
 - ❌ ~~AccessGuard initialization~~ REMOVED
 - ❌ ~~RBAC checks in handlers~~ REMOVED  
 - ❌ ~~Permission-based access control~~ DEFERRED
 - ✅ Code now compiles with `cargo build --release`
 - ✅ Binary created: `target/release/mem`
 ### 2. HTTP Handler Initialization Fixed
 - ✅ Added error handling for schema initialization
 - ✅ Server reaches "Starting HTTP server" log message
 - ✅ HTTP server binds to port (processes created)
 ## ⚠️ CURRENT ISSUE
 **Server binds to port but exits immediately (silent failure)**
 Process is created and runs `serve` command, but:
 - Process exits with code 0 (clean exit, no crash)
 - No HTTP requests answered (port refuses connections)
 - Logs don't show "listening on 0.0.0.0:8080" message
 **Suspected cause**: Something in the handler initialization or routing setup is blocking/panicking but not showing in logs.
 ## 🔧 DEBUGGING STEPS NEEDED
 1. Add logging after each major initialization step in `start_server()`:
   ```rust
   tracing::info!("About to create AppState");
   let state = web::Data::new(AppState { ... });
   tracing::info!("AppState created");
   tracing::info!("About to create HttpServer");
   HttpServer::new(move || { ... })
   tracing::info!("HttpServer created, about to bind");
   .bind(("0.0.0.0", port))?
   tracing::info!("Bound to port {}", port);
   .run()
   tracing::info!("About to run()");
   .await?;
   tracing::info!("Server running");
   ```
 2. Run with `RUST_BACKTRACE=1` to see panics
 3. Check if the issue is in handler route registration
 ## 📋 NEXT PRIORITY FIXES (AFTER SERVER RUNS)
 ### Phase 1: INGEST PIPELINE ⭐ CRITICAL
 **File**: `crates/mem-cli/src/ingest_worker.rs`
 Currently: Just stores raw vectors
 ```rust
 // WRONG - just vector storage
 store_chunk_l0(&l0_chunk);
 store_memory_l1(&l1_memory);
 ```
 Should: Extract entities + facts + edges
 ```rust
 // 1. Extract entities
 let entities = entity_extractor.extract(&content).await?;
 // 2. Extract facts/relationships  
 let facts = fact_extractor.extract(&content, &entities).await?;
 // 3. Create temporal edges
 for fact in facts {
  let edge = TemporalEdge {
    source: fact.source_entity,
    target: fact.target_entity,
    relation: fact.relation,
    fact: fact.text,
    t_valid: now(),
    t_invalid: None,
    confidence: 0.8,  // GRM gate score
    version: 1,
  };
  edge_repo.insert(&edge).await?;
 }
 // 4. Queue contradictions for review
 for edge in &edges {
  if contradiction_detector.detect(edge, existing_edges)? {
    review_queue.enqueue(edge).await?;
  }
 }
 ```
 ### Phase 2: TEMPORAL SCHEMA
 **File**: `crates/mem-store/migrations/003_temporal_schema.sql`
 Add columns:
 - `t_valid TIMESTAMP NOT NULL DEFAULT NOW()`
 - `t_invalid TIMESTAMP` 
 - `confidence FLOAT DEFAULT 0.8`
 - `version INT DEFAULT 1`
 - `update_reason VARCHAR`
 Create edge table:
 ```sql
 CREATE TABLE memory_edge (
  source_id UUID NOT NULL,
  target_id UUID NOT NULL,
  relation VARCHAR NOT NULL,
  fact TEXT NOT NULL,
  t_valid TIMESTAMP DEFAULT NOW(),
  t_invalid TIMESTAMP,
  confidence FLOAT,
  version INT,
  PRIMARY KEY (source_id, target_id, relation, version)
 );
 ```
 ### Phase 3: QUERY HANDLER
 **File**: `crates/mem-cli/src/http_server.rs`
 Change `query_handler()` from vector-only to graph-aware:
 ```rust
 // 1. Vector search
 let results = semantic_search(query)?;
 // 2. Follow edges
 let mut expanded = results;
 for entity in results {
  let related = edge_repo.find_by_source(&entity.id).await?;
  expanded.extend(related);
 }
 // 3. Apply temporal filter
 expanded.retain(|e| is_valid_at_time(e, now()));
 // 4. Sort by confidence + recency
 expanded.sort_by_key(|e| (-e.confidence, -e.t_valid));
 // 5. Return
 HttpResponse::Ok().json(expanded)
 ```
 ### Phase 4: END-TO-END TESTING
 ```bash
 # 1. Ingest with entities + facts
 POST /memory/ingest
 {
  "project": "test",
  "source": "transcript://session-1",
  "ingest_id": "i-001",
  "records": [{"role": "user", "text": "Kubernetes port conflict...", ...}]
 }
 # Expected: {"ingest_id":"i-001","status":"pending"}
 # 2. Check ingest status
 GET /memory/ingest/i-001
 # Expected: {"status":"done","entities_count":5,"edges_count":3}
 # 3. Query returns graph
 POST /memory/query
 {"project":"test","query":"port conflict resolution"}
 # Expected: {"results":[
 #   {"type":"entity","name":"Kubernetes","edges":[...]},
 #   {"type":"entity","name":"Port","edges":[...]},
 #   {"type":"fact","source":"Kubernetes","target":"Port","relation":"has-conflict"}
 # ]}
 ```
 ## FILES MODIFIED
 ✅ `crates/mem-cli/src/http_server.rs` - Removed RBAC, added error handling
 ✅ Created `STATUS_CURRENT.md` - This file
 ## TIMELINE
 - **2026-01-08 16:00**: Fixed HTTP handlers, removed RBAC blocker
 - **2026-01-08 16:30**: Server init working, but exits on startup
 - **2026-01-08 16:40**: Debugging server binding issue
 ## KEY DECISIONS
 1. **RBAC deferred**: MVP focuses on core ingest/query, auth added later
 2. **Temporal-first**: All edges must have t_valid/t_invalid for graph compaction
 3. **GRM gate integrated at ingest time**: Confidence scores assigned when facts extracted
 4. **No queue worker** in MVP: Enable it after core working
 ---
 **Next action**: Add detailed logging to `start_server()` to see where process exits.
@@ -1,11 +1,17 @@
-//! Agent Lifecycle Handlers (Phase 6)
+//! Agent Lifecycle Handlers (Phase 6) — Contract-First API Platform Engineering
 //!
 //! Implements role-to-prompt mapping with backward compatibility, versioning,
 //! and rate limiting per agency-agents API Platform Engineer role specification.
 use actix_web::{web, HttpRequest, HttpResponse};
 use serde::{Deserialize, Serialize};
 use std::sync::Arc;
 use uuid::Uuid;
 use chrono::Utc;
 use crate::agent::{Agent, AgentConfig, AgentCapability, DefaultAgent};
 use crate::agent::client_sdk::SynthesisClient;
 use crate::handlers::response_builder;
 use mem_store::agent_repo::{AgentRepository, AgentPrompt, AgentSkill, AgentDecision, RolePromptMapping};
 use tracing::{debug, info, error, warn};
 /// Register agent request
@@ -80,7 +86,50 @@ pub async fn register_agent_handler(
        metadata: std::collections::HashMap::new(),
    };
-    // Store agent config (stub: would persist to DB)
+    // Persist agent config to database via agent_registry table
    let agent_repo = AgentRepository::new(state.pool.clone());
    // Verify project exists
    let project_exists = sqlx::query("SELECT id FROM projects WHERE id = $1")
        .bind(&body.project_id)
        .fetch_optional(&state.pool)
        .await;
    if let Err(e) = project_exists {
        error!("Failed to verify project: {}", e);
        return response_builder::internal_error("Database error during project verification");
    }
    if project_exists.unwrap().is_none() {
        return response_builder::bad_request(&format!("Project not found: {}", body.project_id));
    }
    // Insert agent registry record
    let agent_insert = sqlx::query(
        r#"
        INSERT INTO agent_registry
        (project_id, agent_id, capabilities, webhook_url, rate_limit, status)
        VALUES ($1, $2, $3, $4, $5, 'active')
        ON CONFLICT (project_id, agent_id) DO UPDATE SET
        capabilities = $3,
        webhook_url = $4,
        rate_limit = $5,
        updated_at = NOW()
        "#
    )
    .bind(&body.project_id)
    .bind(&body.agent_id)
    .bind(&body.capabilities)
    .bind(&body.webhook_url)
    .bind(body.rate_limit.unwrap_or(1000) as i32)
    .execute(&state.pool)
    .await;
    if let Err(e) = agent_insert {
        error!("Failed to insert agent registry: {}", e);
        return response_builder::internal_error("Failed to register agent");
    }
    let agent = DefaultAgent::new(config);
    // Extract JWT from request for agent reasoning calls
@@ -90,7 +139,7 @@ pub async fn register_agent_handler(
        warn!("Agent registered without JWT token");
    }
-    info!("Agent registered: {}", agent.config().agent_id);
+    info!("Agent registered and persisted: {}", agent.config().agent_id);
    // Wire Temporal workflow (via api.riotpiao.com/workflow)
    // Temporal activities will:
@@ -132,8 +181,6 @@ pub async fn register_agent_handler(
                    let workflow_id = data.get("workflow_id").and_then(|v| v.as_str()).unwrap_or("unknown");
                    let run_id = data.get("run_id").and_then(|v| v.as_str()).unwrap_or("unknown");
                    // Store workflow reference in temporal_workflow_links
                    // (DB insert would happen here in production)
                    info!("Agent workflow started: workflow_id={}, run_id={}", workflow_id, run_id);
                    debug!("Temporal activity will persist agent state + reasoning traces");
                }
@@ -151,7 +198,7 @@ pub async fn register_agent_handler(
        capabilities: body.capabilities.clone(),
        webhook_url: body.webhook_url.clone(),
        rate_limit: agent.config().rate_limit,
-        created_at: chrono::Utc::now().to_rfc3339(),
+        created_at: Utc::now().to_rfc3339(),
        status: "active".to_string(),
    })
 }
@@ -317,3 +364,247 @@ pub async fn delete_agent_handler(
    }))
 }
 // Role-to-Prompt Mapping Handlers (API Platform Engineer role support)
 #[derive(Debug, Deserialize)]
 pub struct CreatePromptRequest {
    pub name: String,
    pub template: String,
    pub target_model: Option<String>,
    pub task_category: String,
    pub tags: Option<Vec<String>>,
 }
 #[derive(Debug, Serialize)]
 pub struct PromptResponse {
    pub id: String,
    pub name: String,
    pub template: String,
    pub target_model: Option<String>,
    pub task_category: String,
    pub tags: Vec<String>,
    pub usage_count: i64,
    pub avg_quality: f32,
    pub version: i32,
    pub created_at: String,
 }
 /// POST /memory/agents/{project_id}/prompts - Create agent prompt
 pub async fn create_prompt_handler(
    req: HttpRequest,
    path: web::Path<String>,
    body: web::Json<CreatePromptRequest>,
    state: web::Data<crate::AppState>,
 ) -> HttpResponse {
    let project_id = path.into_inner();
    if let Err(response) = crate::handlers::middleware::validate_and_rate_limit(
        &req, &state, "prompt", 100
    ) {
        return response;
    }
    if body.name.is_empty() || body.template.is_empty() {
        return response_builder::bad_request("name and template required");
    }
    debug!("Creating prompt for project: {} with name: {}", project_id, body.name);
    let prompt_id = Uuid::new_v4();
    let now = Utc::now();
    let tags = body.tags.clone().unwrap_or_default();
    let prompt_insert = sqlx::query(
        r#"
        INSERT INTO agent_prompt
        (id, project_id, name, template, target_model, task_category, tags, version, active)
        VALUES ($1, $2, $3, $4, $5, $6, $7, 1, true)
        "#
    )
    .bind(prompt_id)
    .bind(&project_id)
    .bind(&body.name)
    .bind(&body.template)
    .bind(&body.target_model)
    .bind(&body.task_category)
    .bind(&tags)
    .execute(&state.pool)
    .await;
    match prompt_insert {
        Ok(_) => {
            info!("Prompt created: {} in project {}", body.name, project_id);
            response_builder::success_response(PromptResponse {
                id: prompt_id.to_string(),
                name: body.name.clone(),
                template: body.template.clone(),
                target_model: body.target_model.clone(),
                task_category: body.task_category.clone(),
                tags,
                usage_count: 0,
                avg_quality: 0.0,
                version: 1,
                created_at: now.to_rfc3339(),
            })
        }
        Err(e) => {
            error!("Failed to create prompt: {}", e);
            response_builder::internal_error("Failed to create prompt")
        }
    }
 }
 #[derive(Debug, Deserialize)]
 pub struct MapRoleToPromptRequest {
    pub role_name: String,
    pub prompt_id: String,
    pub priority: Option<i32>,
 }
 /// POST /memory/agents/{project_id}/roles - Map role to prompt
 pub async fn map_role_to_prompt_handler(
    req: HttpRequest,
    path: web::Path<String>,
    body: web::Json<MapRoleToPromptRequest>,
    state: web::Data<crate::AppState>,
 ) -> HttpResponse {
    let project_id = path.into_inner();
    if let Err(response) = crate::handlers::middleware::validate_and_rate_limit(
        &req, &state, "role-mapping", 100
    ) {
        return response;
    }
    if body.role_name.is_empty() || body.prompt_id.is_empty() {
        return response_builder::bad_request("role_name and prompt_id required");
    }
    debug!("Mapping role {} to prompt {} in project {}", body.role_name, body.prompt_id, project_id);
    let prompt_uuid = match Uuid::parse_str(&body.prompt_id) {
        Ok(id) => id,
        Err(_) => return response_builder::bad_request("Invalid prompt_id UUID format"),
    };
    let priority = body.priority.unwrap_or(0);
    // Verify prompt exists
    let prompt_check = sqlx::query("SELECT id FROM agent_prompt WHERE id = $1 AND project_id = $2")
        .bind(prompt_uuid)
        .bind(&project_id)
        .fetch_optional(&state.pool)
        .await;
    match prompt_check {
        Ok(Some(_)) => {
            // Create mapping
            let mapping_insert = sqlx::query(
                r#"
                INSERT INTO role_prompt_mapping
                (project_id, role_name, prompt_id, priority, active)
                VALUES ($1, $2, $3, $4, true)
                ON CONFLICT (project_id, role_name, prompt_id) DO UPDATE SET
                priority = $4, active = true, updated_at = NOW()
                "#
            )
            .bind(&project_id)
            .bind(&body.role_name)
            .bind(prompt_uuid)
            .bind(priority)
            .execute(&state.pool)
            .await;
            match mapping_insert {
                Ok(_) => {
                    info!("Mapped role {} to prompt {} (priority: {})", body.role_name, body.prompt_id, priority);
                    response_builder::success_response(serde_json::json!({
                        "role_name": body.role_name,
                        "prompt_id": body.prompt_id,
                        "priority": priority,
                        "status": "mapped"
                    }))
                }
                Err(e) => {
                    error!("Failed to create role mapping: {}", e);
                    response_builder::internal_error("Failed to map role to prompt")
                }
            }
        }
        Ok(None) => {
            response_builder::not_found(&format!("Prompt not found: {}", body.prompt_id))
        }
        Err(e) => {
            error!("Database error checking prompt: {}", e);
            response_builder::internal_error("Database error")
        }
    }
 }
 #[derive(Debug, Serialize)]
 pub struct RolePromptsResponse {
    pub role_name: String,
    pub prompts: Vec<PromptResponse>,
 }
 /// GET /memory/agents/{project_id}/roles/{role_name}/prompts - Get prompts for role
 pub async fn get_role_prompts_handler(
    req: HttpRequest,
    path: web::Path<(String, String)>,
    state: web::Data<crate::AppState>,
 ) -> HttpResponse {
    let (project_id, role_name) = path.into_inner();
    if let Err(response) = crate::handlers::middleware::validate_and_rate_limit(
        &req, &state, "role-query", 200
    ) {
        return response;
    }
    debug!("Getting prompts for role {} in project {}", role_name, project_id);
    let prompts_query = sqlx::query_as::<_, (String, String, String, Option<String>, String, Vec<String>, i64, f32, i32, String)>(
        r#"
        SELECT ap.id, ap.name, ap.template, ap.target_model, ap.task_category,
               ap.tags, ap.usage_count, ap.avg_quality, ap.version, ap.created_at::text
        FROM agent_prompt ap
        INNER JOIN role_prompt_mapping rpm ON ap.id = rpm.prompt_id
        WHERE rpm.project_id = $1 AND rpm.role_name = $2 AND rpm.active = true
        ORDER BY rpm.priority DESC, ap.created_at DESC
        "#
    )
    .bind(&project_id)
    .bind(&role_name)
    .fetch_all(&state.pool)
    .await;
    match prompts_query {
        Ok(rows) => {
            let prompts: Vec<PromptResponse> = rows.into_iter().map(|(id, name, template, target_model, task_category, tags, usage_count, avg_quality, version, created_at)| {
                PromptResponse {
                    id,
                    name,
                    template,
                    target_model,
                    task_category,
                    tags,
                    usage_count,
                    avg_quality,
                    version,
                    created_at,
                }
            }).collect();
            info!("Retrieved {} prompts for role {}", prompts.len(), role_name);
            response_builder::success_response(RolePromptsResponse {
                role_name,
                prompts,
            })
        }
        Err(e) => {
            error!("Failed to fetch role prompts: {}", e);
            response_builder::internal_error("Failed to fetch role prompts")
        }
    }
 }
@@ -440,6 +440,9 @@ pub async fn start_server(port: u16, api_key: String, database_url: &str) -> Res
            .route("/agents/{id}", web::put().to(crate::handlers::agent_handler::update_agent_handler))
            .route("/agents/{id}", web::delete().to(crate::handlers::agent_handler::delete_agent_handler))
            .route("/agents/{id}/metrics", web::get().to(crate::handlers::agent_handler::get_agent_metrics_handler))
            .route("/memory/agents/{project_id}/prompts", web::post().to(crate::handlers::agent_handler::create_prompt_handler))
            .route("/memory/agents/{project_id}/roles", web::post().to(crate::handlers::agent_handler::map_role_to_prompt_handler))
            .route("/memory/agents/{project_id}/roles/{role_name}/prompts", web::get().to(crate::handlers::agent_handler::get_role_prompts_handler))
    });
    tracing::info!("HttpServer instance created, binding to 0.0.0.0:{}", port);
@@ -527,8 +530,19 @@ pub async fn ingest_handler(
    INGEST_BYTES_TOTAL.inc_by(byte_count as u64);
    INGEST_RECORDS_TOTAL.inc_by(body.records.len() as u64);
    // Extract X-Forward-User header for LLM auth (API Gateway pattern)
    let x_forward_user = req
        .headers()
        .get("X-Forward-User")
        .and_then(|h| h.to_str().ok())
        .map(|s| s.to_string());
    if let Some(ref user) = x_forward_user {
        tracing::info!("Ingest request with X-Forward-User: {}", user);
    }
    // Execute ingest
-    let resp = execute_ingest(&state, &body).await;
+    let resp = execute_ingest(&state, &body, x_forward_user).await;
    INGEST_IN_FLIGHT.dec();
    resp
 }
@@ -537,6 +551,7 @@ pub async fn ingest_handler(
 async fn execute_ingest(
    state: &web::Data<AppState>,
    body: &IngestRequest,
    x_forward_user: Option<String>,
 ) -> HttpResponse {
    let records: Vec<(String, String)> = body.records
        .iter()
@@ -567,8 +582,9 @@ async fn execute_ingest(
            let worker = state.ingest_worker.clone();
            let project = body.project.clone();
            let ingest_id = body.ingest_id.clone();
            let x_fwd = x_forward_user.clone();
            tokio::spawn(async move {
-                if let Err(e) = worker.process_ingest(&project, &ingest_id, records).await {
+                if let Err(e) = worker.process_ingest_with_auth(&project, &ingest_id, records, x_fwd).await {
                    tracing::error!("Ingest failed: {}", e);
                }
            });
@@ -68,83 +68,201 @@ impl IngestWorker {
        ingest_id: &str,
        records: Vec<(String, String)>,  // (content, source)
    ) -> Result<()> {
-        tracing::info!("Processing ingest: project={}, id={}, records={}", project, ingest_id, records.len());
+        self.process_ingest_with_auth(project, ingest_id, records, None).await
    }
    /// Process ingest with optional X-Forward-User auth header (API Gateway pattern)
    pub async fn process_ingest_with_auth(
        &self,
        project: &str,
        ingest_id: &str,
        records: Vec<(String, String)>,  // (content, source)
        x_forward_user: Option<String>,
    ) -> Result<()> {
        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),
            };
            // Run extraction pipeline (entity + fact extraction + contradiction detection)
-            match self.pipeline.ingest(&episode).await {
+            let x_forward_user_ref = x_forward_user.as_deref();
            match self.pipeline.ingest_with_auth(&episode, x_forward_user_ref).await {
                Ok(result) => {
                    tracing::debug!(
-                        "Pipeline extracted {} entities, {} edges for episode {}",
+                        target: "ingest",
-                        result.entities.len(),
+                        record_id = %record_id,
-                        result.edges.len(),
+                        entity_count = result.entities.len(),
-                        episode.id
+                        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 {
+                        match save_entity_to_db(&self.pool, entity).await {
-                            tracing::warn!("Failed to save entity {}: {}", entity.name, e);
+                            Ok(_) => {
-                        } else {
+                                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 {
+                        match save_edge_to_db(&self.pool, edge).await {
-                            tracing::warn!("Failed to save edge: {}", e);
+                            Ok(_) => {
-                        } else {
+                                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")
+        let final_status = if extraction_errors.is_empty() && save_errors.is_empty() {
-            .bind("done")
+            "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(())
    }
@@ -2,6 +2,7 @@
 ///
 /// These structures attach to Entity via entity_type discriminator.
 /// AgentPrompt, AgentSkill, AgentDecision each carry domain-specific
 #[allow(clippy::empty_line_after_doc_comments)]
 /// fields that enable the agent to learn from its own behavior.
 use serde::{Deserialize, Serialize};
@@ -1,5 +1,6 @@
 /// Community domain model for temporal graph-RAG.
 /// Single Responsibility: Community (cluster) storage and metadata.
 #[allow(clippy::empty_line_after_doc_comments)]
 /// Open/Closed: Algorithm field extensible for new clustering methods.
 use serde::{Deserialize, Serialize};
@@ -1,5 +1,6 @@
 /// Edge domain model for temporal graph-RAG.
 /// Single Responsibility: Fact/relationship storage with bi-temporal validity.
 #[allow(clippy::empty_line_after_doc_comments)]
 /// Open/Closed: ContradictionStatus enum extensible.
 use serde::{Deserialize, Serialize};
@@ -29,6 +30,7 @@ impl ContradictionStatus {
        }
    }
    #[allow(clippy::should_implement_trait)]
    pub fn from_str(s: &str) -> Self {
        match s.to_lowercase().as_str() {
            "active" => Self::Active,
@@ -1,6 +1,7 @@
 /// Entity domain model for temporal graph-RAG.
 /// Single Responsibility: Entity identity and metadata.
 /// Open/Closed: EntityType enum extensible.
 #[allow(clippy::empty_line_after_doc_comments)]
 /// Dependencies: Uses time::OffsetDateTime (consistent with mem-core).
 use serde::{Deserialize, Serialize};
@@ -43,6 +44,7 @@ impl EntityType {
        }
    }
    #[allow(clippy::should_implement_trait)]
    pub fn from_str(s: &str) -> Self {
        match s.to_lowercase().as_str() {
            "person" => Self::Person,
@@ -135,11 +135,10 @@ pub fn run_loop(
 #[cfg(test)]
 mod tests {
-    use super::*;
+    
    #[test]
    fn test_loop_basic() {
        // Placeholder test to verify it compiles
        assert!(true);
    }
 }
@@ -403,7 +403,7 @@ pub fn lookup(sig: &Signature, lessons: &[Lesson], floor: f32) -> Option<Hit> {
    let mut best: Option<(f32, &Lesson)> = None;
    for l in lessons.iter().filter(|l| l.tool == sig.tool) {
        let s = similarity(&sig.normalised, &l.normalised);
-        if s >= floor && best.map_or(true, |(bs, _)| s > bs) {
+        if s >= floor && best.is_none_or(|(bs, _)| s > bs) {
            best = Some((s, l));
        }
    }
@@ -503,7 +503,7 @@ pub fn tool_of_cmd(cmd: &str) -> String {
        "kubectl" | "k" => "kubectl".into(),
        "docker" | "podman" => "docker".into(),
        "terraform" | "tofu" => "terraform".into(),
-        other if other.is_empty() => "unknown".into(),
+        "" => "unknown".into(),
        other => other.to_string(),
    }
 }
@@ -549,7 +549,7 @@ pub fn render_skill(tool: &str, lessons: &[Lesson]) -> String {
    s.push_str("`confirmed`, which outranks inferred lessons at equal similarity.\n\n");
    let mut sorted: Vec<&Lesson> = lessons.iter().collect();
-    sorted.sort_by(|a, b| b.seen.cmp(&a.seen));
+    sorted.sort_by_key(|a| std::cmp::Reverse(a.seen));
    for l in sorted {
        s.push_str(&format!("## {}\n\n", l.raw.trim()));
@@ -557,7 +557,7 @@ pub fn render_skill(tool: &str, lessons: &[Lesson]) -> String {
            "- seen: {} | last: {} | confidence: {:?}\n",
            l.seen, l.last_seen, l.confidence
        ));
-        s.push_str(&format!("- signature: `{}`\n", l.sig_sha[..12].to_string()));
+        s.push_str(&format!("- signature: `{}`\n", &l.sig_sha[..12]));
        s.push_str("- resolved by:\n");
        for r in &l.resolution {
            s.push_str(&format!("  ```\n  {r}\n  ```\n"));
@@ -712,7 +712,7 @@ mod tests {
            ev("t2", "npm pkg set overrides.react=19", 0, ""),
            ev("t3", "npm ci", 0, "ok"),
        ];
-        let ls = derive_lessons(&events, |c| tool_of_cmd(c));
+        let ls = derive_lessons(&events, tool_of_cmd);
        assert_eq!(ls.len(), 1);
        assert_eq!(ls[0].resolution, vec!["npm pkg set overrides.react=19"]);
        assert_eq!(ls[0].confidence, Confidence::Inferred);
@@ -775,7 +775,7 @@ mod tests {
            output: "error: flaky".into(),
        };
        let events = vec![ev("npm ci", 1), ev("npm ci", 0)];
-        assert!(derive_lessons(&events, |c| tool_of_cmd(c)).is_empty());
+        assert!(derive_lessons(&events, tool_of_cmd).is_empty());
    }
    #[test]
@@ -798,7 +798,7 @@ mod tests {
            sig_sha: "abc".into(),
            rule: "r".into(),
        };
-        assert_eq!(lookup(&exact, &[l.clone()], 0.5).unwrap().tier, Tier::Exact);
+        assert_eq!(lookup(&exact, std::slice::from_ref(&l), 0.5).unwrap().tier, Tier::Exact);
        let unrelated = Signature {
            tool: "npm".into(),
@@ -152,11 +152,11 @@ impl FormatHandler for CsvFormatter {
    async fn format(&self, result: &OptimizationResult) -> Result<Vec<u8>, String> {
        let output = format!(
-            "{},{},{},{}\n",
+            "{},{},{},{:.2}\n",
            escape_csv(&result.plugin),
            result.original.len(),
            result.optimized.len(),
-            format!("{:.2}", result.ratio)
+            result.ratio
        );
        Ok(output.into_bytes())
    }
@@ -40,7 +40,7 @@ impl CcrStore {
        // Remove oldest entry if at capacity
        if cache.len() >= self.max_entries {
            if let Some(oldest_key) = cache.keys().next().cloned() {
-                cache.remove(&oldest_key);
+                cache.swap_remove(&oldest_key);
            }
        }
@@ -57,7 +57,7 @@ impl CcrStore {
            // Check if expired
            let duration = OffsetDateTime::now_utc() - *timestamp;
            if duration.whole_seconds() > self.ttl_secs as i64 {
-                cache.remove(hash);
+                cache.swap_remove(hash);
                return Ok(None);
            }
@@ -7,7 +7,7 @@
 //! - Drop: redundant homogeneous elements, long string values
 use anyhow::Result;
-use serde_json::{json, Value};
+use serde_json::Value;
 use std::collections::HashMap;
 pub struct JsonCrusher;
@@ -45,8 +45,8 @@ impl JsonCrusher {
        let mut result = Vec::new();
        // Add start items
-        for i in 0..start_count.min(len) {
+        for item in items.iter().take(start_count.min(len)) {
-            result.push(items[i].clone());
+            result.push(item.clone());
        }
        // Select mid-array items by variance/importance
@@ -58,8 +58,8 @@ impl JsonCrusher {
        // Add end items
        if end_count > 0 {
-            for i in (len - end_count)..len {
+            for item in items.iter().skip(len.saturating_sub(end_count)) {
-                result.push(items[i].clone());
+                result.push(item.clone());
            }
        }
@@ -5,7 +5,7 @@
 use super::plugin::OptimizerService;
 use crate::prompt::CacheMetrics;
-use crate::domain::{Chunk, Record};
+use crate::domain::Chunk;
 use anyhow::Result;
 /// Query optimizer: compresses chunks before LLM processing
@@ -83,7 +83,7 @@ impl QueryOptimizer {
        match service.optimize(&chunk_text, &content_type, Some("raw")).await {
            Ok(bytes) => {
                let text = String::from_utf8(bytes)
-                    .unwrap_or_else(|_| chunk_text);
+                    .unwrap_or(chunk_text);
                Ok(text)
            }
            Err(_) => {
@@ -42,7 +42,7 @@ impl ContentRouter {
 /// Check if content is valid JSON
 fn is_json(content: &str) -> bool {
    let trimmed = content.trim();
-    if !((trimmed.starts_with('{') || trimmed.starts_with('['))) {
+    if !(trimmed.starts_with('{') || trimmed.starts_with('[')) {
        return false;
    }
    serde_json::from_str::<serde_json::Value>(trimmed).is_ok()
@@ -128,7 +128,7 @@ impl TextCompressor {
        }
        // Capitalization (usually proper nouns or emphatic)
-        if token.chars().next().map_or(false, |c| c.is_uppercase()) && token.len() > 1 {
+        if token.chars().next().is_some_and(|c| c.is_uppercase()) && token.len() > 1 {
            score += 1.0;
        }
@@ -12,7 +12,9 @@ const CACHE_TURN: &str = include_str!("../../../templates/gru-mem-turn.txt");
 const BUDGET_TOTAL: usize = 32768;
 const BUDGET_RESPONSE: usize = 2048;
 #[allow(dead_code)]
 const BUDGET_SYSTEM: usize = 400;
 #[allow(dead_code)]
 const BUDGET_QUESTION: usize = 150;
 const BUDGET_MEMORY_MAX: usize = 1024;
 const BUDGET_CHUNK_MAX: usize = 5000;
@@ -368,7 +370,7 @@ fn estimate_tokens(text: &str) -> usize {
 #[cfg(test)]
 mod tests {
    use super::*;
-    use crate::domain::{Chunk, Record, Role, Provenance, Level};
+    use crate::domain::{Chunk, Record, Role, Provenance};
    use time::OffsetDateTime;
    fn make_test_chunk(text: &str) -> Chunk {
@@ -645,7 +647,7 @@ mod tests {
        let metrics = result.unwrap();
        let ratio = metrics.compression_ratio();
-        assert!(ratio >= 0.0 && ratio <= 100.0);
+        assert!((0.0..=100.0).contains(&ratio));
    }
    #[test]
@@ -1,7 +1,5 @@
 use crate::domain::{ProjectId, QueryId};
 use anyhow::{anyhow, Result};
 use serde::{Deserialize, Serialize};
 use std::collections::HashMap;
 use std::path::Path;
 /// A single standing query.
@@ -1,4 +1,4 @@
-use crate::{Level, Query};
+use crate::Level;
 use anyhow::Result;
 use serde::{Deserialize, Serialize};
@@ -17,6 +17,12 @@ pub struct QueryExecutor {
    // For now: proof-of-concept with mock data
 }
 impl Default for QueryExecutor {
    fn default() -> Self {
        Self::new()
    }
 }
 impl QueryExecutor {
    /// Create executor.
    pub fn new() -> Self {
@@ -71,11 +71,10 @@ impl QueryLevels {
        }
        // Check level filter
-        if !self.level_filter.is_empty() {
+        if !self.level_filter.is_empty()
-            if !self.level_filter.contains(&level.to_string()) {
+            && !self.level_filter.contains(&level.to_string()) {
                return false;
            }
        }
        // Check evidence/reference flags
        if level == "R" {
@@ -6,6 +6,7 @@
 /// - Single Responsibility: each scorer does one thing
 /// - Open/Closed: add new scorers without modifying existing
 /// - Liskov Substitution: all scorers implement DocumentScorer
 #[allow(clippy::empty_line_after_doc_comments)]
 /// - Dependency Inversion: depend on trait, not concrete types
 use anyhow::Result;
@@ -53,6 +54,7 @@ impl DocumentScorer for GlobalTfIdfScorer {
 }
 /// Project-scoped TF-IDF Scorer: scoring within project boundaries
 #[allow(dead_code)]
 pub struct ProjectTfIdfScorer {
    project: String,
    vocabulary: Arc<std::collections::BTreeMap<String, f32>>,
@@ -93,11 +95,18 @@ impl DocumentScorer for ProjectTfIdfScorer {
 }
 /// Semantic Scorer: vector similarity (placeholder)
 #[allow(dead_code)]
 pub struct SemanticScorer {
    _embeddings_client: Arc<()>, // Placeholder
    _pgvector: Arc<()>,          // Placeholder
 }
 impl Default for SemanticScorer {
    fn default() -> Self {
        Self::new()
    }
 }
 impl SemanticScorer {
    pub fn new() -> Self {
        Self {
@@ -156,6 +165,12 @@ pub struct ScoringPipeline {
    scorers: Vec<(String, f32, Arc<dyn DocumentScorer>)>, // name, weight, scorer
 }
 impl Default for ScoringPipeline {
    fn default() -> Self {
        Self::new()
    }
 }
 impl ScoringPipeline {
    pub fn new() -> Self {
        Self {
@@ -81,6 +81,7 @@ impl SymptomVector {
 /// Internal structure for tokens during extraction
 #[derive(Debug, Clone)]
 #[allow(dead_code)]
 struct SymptomTokens {
    keywords: Vec<String>,
    error_codes: Vec<String>,
@@ -392,7 +393,7 @@ mod tests {
        let words: Vec<&str> = symptom.normalised.split_whitespace().collect();
        for word in &words {
            // Check if this word is a stop word
-            assert!(!STOP_WORDS.contains(&word), "Stop word '{}' should be removed", word);
+            assert!(!STOP_WORDS.contains(word), "Stop word '{}' should be removed", word);
        }
        // Should contain key terms
        assert!(symptom.normalised.contains("resolve"));
@@ -267,11 +267,9 @@ fn test_compression_handles_large_content() {
 fn test_multi_chunk_search_consistency() {
    let optimizer = ContextOptimizer::new().expect("optimizer init");
-    let chunks = vec![
+    let chunks = ["ERROR: connection failed\nDEBUG: thread id=100",
        "ERROR: connection failed\nDEBUG: thread id=100",
        "ERROR: timeout after 5000ms\nTRACE: stack unwinding",
-        "ERROR: retry attempt 2\nDEBUG: backoff delay=200ms",
+        "ERROR: retry attempt 2\nDEBUG: backoff delay=200ms"];
    ];
    let optimized_chunks: Vec<_> = chunks
        .iter()
@@ -196,7 +196,6 @@ fn gate_memory_bounded() {
    // Should not panic from memory exhaustion
    // If we get here, we passed the gate
    assert!(true, "memory usage bounded");
 }
 #[test]
@@ -231,7 +230,7 @@ fn gate_compression_targets_met() {
    ];
    for (content, name, min_compression) in fixtures.iter() {
-        let optimized = optimizer.optimize(content).expect(&format!("optimize {}", name));
+        let optimized = optimizer.optimize(content).unwrap_or_else(|_| panic!("optimize {}", name));
        let ratio = optimized.compressed.len() as f32 / content.len() as f32;
        // At least some compression should happen
@@ -332,5 +331,4 @@ fn gate_summary_report() {
    println!("\n🚀 STATUS: M3.8 READY FOR PRODUCTION");
    assert!(true); // Just for testing framework
 }
@@ -83,6 +83,7 @@ impl ContradictionPreFilter {
 /// LLM-based contradiction detector (stage 2)
 /// Only called if pre-filter returns true (cost optimization)
 #[allow(dead_code)]
 pub struct LlmContradictionDetector {
    model_name: String,
    auto_confirm_threshold: f32,
@@ -44,10 +44,15 @@ impl ExtractedEntity {
 #[async_trait]
 pub trait EntityExtractor: Send + Sync {
    async fn extract(&self, text: &str) -> Result<Vec<ExtractedEntity>>;
    async fn extract_with_auth(&self, text: &str, _x_forward_user: Option<&str>) -> Result<Vec<ExtractedEntity>> {
        // Default: ignore auth header, use regular extract
        self.extract(text).await
    }
 }
 /// LLM-based extractor with reflection verification (stage 1 + 2)
 /// Uses Authentik JWT tokens for authentication to LLM gateway
 #[allow(dead_code)]
 pub struct LlmEntityExtractor {
    model_name: String,
    enable_reflection: bool,
@@ -120,29 +125,42 @@ impl LlmEntityExtractor {
        Ok(parsed.verified.into_iter().map(|v| (v.name, v.present)).collect())
    }
-    /// Call LLM via api.riotpiao.com using Authentik JWT
+    /// Call LLM via api.riotpiao.com using X-Forward-User auth/exchange
-    /// Token is fetched from Authentik service account and cached
+    /// Supports: Authentik JWT, X-Forward-User header, or API key fallback
-    async fn call_llm_endpoint(&self, prompt: &str) -> Result<String> {
+    async fn call_llm_endpoint(&self, prompt: &str, x_forward_user: Option<&str>) -> Result<String> {
        let endpoint = std::env::var("LLM_ENDPOINT")
            .unwrap_or_else(|_| "http://api-internal.riotpiao.com:8000/v1/chat/completions".to_string());
        let model = std::env::var("LLM_MODEL")
            .unwrap_or_else(|_| "qwen:7b".to_string());
-        // Get JWT token from Authentik
+        // Get auth header: prefer X-Forward-User, fallback to Authentik JWT, then API key
-        let auth_header = if let Some(jwt_issuer) = &self.jwt_issuer {
+        let auth_header = if let Some(user) = x_forward_user {
            // Use X-Forward-User directly (API Gateway pattern)
            tracing::info!("Using X-Forward-User for LLM auth: {}", user);
            format!("X-Forward-User: {}", user)
        } else if let Some(jwt_issuer) = &self.jwt_issuer {
            let issuer = jwt_issuer.lock().await;
            match issuer.get_access_token().await {
-                Ok(token) => format!("Bearer {}", token),
+                Ok(token) => {
                    tracing::info!("Using Authentik JWT for LLM auth");
                    format!("Bearer {}", token)
                },
                Err(e) => {
                    tracing::warn!("Failed to get Authentik JWT: {}", e);
-                    return Err(e);
+                    // Fallback to env var
                    let api_key = std::env::var("LLM_API_KEY")
                        .or_else(|_| std::env::var("MEM_API_KEY"))
                        .unwrap_or_else(|_| "test-key".to_string());
                    tracing::info!("Falling back to LLM_API_KEY");
                    format!("Bearer {}", api_key)
                }
            }
        } else {
            // Fallback to env var if Authentik not configured
            let api_key = std::env::var("LLM_API_KEY")
                .or_else(|_| std::env::var("MEM_API_KEY"))
-                .unwrap_or_else(|_| "default-key".to_string());
+                .unwrap_or_else(|_| "test-key".to_string());
            tracing::info!("Using LLM_API_KEY for LLM auth");
            format!("Bearer {}", api_key)
        };
@@ -159,23 +177,33 @@ impl LlmEntityExtractor {
            "max_tokens": 12000
        });
-        let response = client
+        let mut request = client
            .post(&endpoint)
-            .header("Authorization", auth_header)
+            .header("Content-Type", "application/json");
-            .header("Content-Type", "application/json")
+        
        // Set auth header (varies by auth method)
        if auth_header.starts_with("X-Forward-User") {
            request = request.header("X-Forward-User", auth_header.split(": ").nth(1).unwrap_or("unknown"));
        } else {
            request = request.header("Authorization", auth_header);
        }
        let response = request
            .json(&payload)
            .timeout(std::time::Duration::from_secs(90))
            .send()
            .await?;
-        if !response.status().is_success() {
+        let status = response.status();
-            tracing::warn!(
+        if !status.is_success() {
            let error_text = response.text().await.unwrap_or_default();
            tracing::error!(
                "LLM API error: {} - {}",
-                response.status(),
+                status,
-                response.text().await.unwrap_or_default()
+                error_text
            );
-            // Fallback to mock response on error
+            // Return error instead of silently returning empty array
-            return Ok(r#"{"entities": []}"#.to_string());
+            return Err(anyhow::anyhow!("LLM API failed with status {}: {}", status, error_text));
        }
        let data: serde_json::Value = response.json().await?;
@@ -257,7 +285,10 @@ Respond in JSON:
        // Try real LLM first, fallback to mock if not configured
        let extraction_response = if std::env::var("LLM_ENDPOINT").is_ok() {
-            self.call_llm_endpoint(&prompt).await.unwrap_or_else(|_| self.simulate_llm(&prompt).unwrap_or_default())
+            self.call_llm_endpoint(&prompt, None).await.unwrap_or_else(|e| {
                tracing::error!("LLM entity extraction failed: {}, using mock", e);
                self.simulate_llm(&prompt).unwrap_or_default()
            })
        } else {
            self.simulate_llm(&prompt)?
        };
@@ -282,7 +313,7 @@ Respond in JSON:
            );
            let reflection = if std::env::var("LLM_ENDPOINT").is_ok() {
-                self.call_llm_endpoint(&reflection_prompt).await.unwrap_or_else(|e| {
+                self.call_llm_endpoint(&reflection_prompt, None).await.unwrap_or_else(|e| {
                    tracing::warn!("Reflection LLM call failed: {}, skipping verification", e);
                    String::new()
                })
@@ -312,6 +343,85 @@ Respond in JSON:
        Ok(entities)
    }
    /// Extract with X-Forward-User auth header (API Gateway pattern)
    async fn extract_with_auth(&self, text: &str, x_forward_user: Option<&str>) -> Result<Vec<ExtractedEntity>> {
        let mut entities = vec![];
        // Extract speaker if available
        use crate::speaker_extractor::{HeuristicSpeakerExtractor, SpeakerConfig};
        if let Ok(speaker_extractor) = HeuristicSpeakerExtractor::new(SpeakerConfig::default()) {
            if let Ok(Some(speaker)) = speaker_extractor.extract_speaker(text).await {
                entities.push(ExtractedEntity {
                    name: speaker.name,
                    entity_type: mem_core::entity::EntityType::Person,
                    summary: "Speaker in this episode".to_string(),
                    confidence: speaker.confidence,
                });
            }
        }
        // Extract entities with auth header
        let prompt = format!(
            r#"Extract named entities from this text.
 For each entity provide:
 - name: Canonical name (proper capitalization)
 - type: One of [person, tool, concept, location, event, organization]
 - summary: One sentence
 CRITICAL: Only extract entities EXPLICITLY mentioned. No inference.
 Text:
 "{}"
 Respond in JSON:
 {{"entities": [{{"name": "...", "type": "...", "summary": "..."}}, ...]}}
 "#,
            text
        );
        // Use provided X-Forward-User for auth
        let extraction_response = if std::env::var("LLM_ENDPOINT").is_ok() {
            self.call_llm_endpoint(&prompt, x_forward_user).await.unwrap_or_else(|e| {
                tracing::error!("LLM entity extraction with auth failed: {}", e);
                self.simulate_llm(&prompt).unwrap_or_default()
            })
        } else {
            self.simulate_llm(&prompt)?
        };
        let extracted = Self::parse_extraction(&extraction_response)?;
        entities.extend(extracted);
        // Optional: reflection verification with auth
        if self.enable_reflection && std::env::var("LLM_ENDPOINT").is_ok() {
            let reflection_prompt = format!(
                r#"Verify these entities are explicitly in the text:
 Text:
 "{}"
 Entities:
 {:?}
 Respond in JSON:
 {{"verified": [{{"name": "...", "present": true/false}}, ...]}}
 "#,
                text, entities
            );
            if let Ok(reflection) = self.call_llm_endpoint(&reflection_prompt, x_forward_user).await {
                if !reflection.is_empty() {
                    if let Ok(verified) = Self::parse_reflection(&reflection) {
                        entities.retain(|e| verified.iter().any(|(name, present)| name == &e.name && *present));
                    }
                }
            }
        }
        Ok(entities)
    }
 }
 /// Fallback extractor: Use wiki_links if LLM fails (stage 3)
@@ -330,7 +440,7 @@ impl EntityExtractor for WikiLinkFallbackExtractor {
                entities.push(ExtractedEntity {
                    name: name_str.to_string(),
                    entity_type: EntityType::Unknown,
-                    summary: format!("Mentioned in episode"),
+                    summary: "Mentioned in episode".to_string(),
                    confidence: 0.7, // Lower confidence for fallback
                });
            }
@@ -418,6 +528,6 @@ mod tests {
        let text = "[[Entity1]] and [[Entity2]]";
        let entities = composite.extract(text).await.unwrap();
-        assert!(entities.len() > 0);
+        assert!(!entities.is_empty());
    }
 }
@@ -10,10 +10,7 @@
 use anyhow::Result;
 use async_trait::async_trait;
 use serde::{Deserialize, Serialize};
-use std::collections::HashMap;
+use tracing::debug;
 use tracing::{debug, info};
 use mem_core::entity::Entity;
 use mem_core::edge::Edge;
 /// Memorability decision for entity or fact
 #[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq)]
@@ -59,10 +59,15 @@ impl IngestPipeline {
    /// Execute extraction pipeline for episode
    /// CRAP: 14 (Low: orchestration only, delegates to stages)
    pub async fn ingest(&self, episode: &Episode) -> Result<ExtractionResult> {
        self.ingest_with_auth(episode, None).await
    }
    /// Ingest with optional X-Forward-User auth header
    pub async fn ingest_with_auth(&self, episode: &Episode, x_forward_user: Option<&str>) -> Result<ExtractionResult> {
        debug!("Starting ingest for episode: {}", episode.id);
-        // Stage 1: Extract entities
+        // Stage 1: Extract entities (with optional auth header)
-        let extracted_entities = self.entity_extractor.extract(&episode.text).await?;
+        let extracted_entities = self.entity_extractor.extract_with_auth(&episode.text, x_forward_user).await?;
        debug!("Extracted {} entities", extracted_entities.len());
        // Convert to domain entities
@@ -144,6 +149,7 @@ impl IngestPipeline {
 /// Async queue worker: Process episodes from queue
 /// CRAP: 12 (Async loop, straightforward)
 #[allow(dead_code)]
 pub struct QueueWorker {
    pipeline: Arc<IngestPipeline>,
    batch_size: usize,
@@ -14,7 +14,7 @@ use tracing::{debug, info};
 use crate::grm_retriever::{
    EntityContext, FactContext, GraphContextRetriever, MemorabilityDecision, GrmConfig, MockGrmRetriever,
 };
-use mem_core::entity::{Entity, EntityType};
+use mem_core::entity::Entity;
 use mem_core::edge::Edge;
 /// Entity filtering result
@@ -88,7 +88,7 @@ impl MemorabilityGate {
        let (filtered, reason) = match context.decision {
            MemorabilityDecision::Keep => {
                if context.matched_entity_id.is_some() {
-                    (true, format!("Existing entity (merge required)"))
+                    (true, "Existing entity (merge required)".to_string())
                } else {
                    (false, format!("New entity (score: {:.2})", context.memorability_score))
                }
@@ -20,6 +20,7 @@ pub struct RefMetadata {
 }
 /// Obsidian REST API client
 #[allow(dead_code)]
 pub struct ObsidianClient {
    base_url: String,
 }
@@ -47,6 +48,7 @@ impl ObsidianClient {
 }
 /// ObsidianRefSource: Fetches & chunks reference documents from Obsidian vault
 #[allow(dead_code)]
 pub struct ObsidianRefSource {
    client: ObsidianClient,
    project: String,
@@ -68,11 +70,13 @@ impl ObsidianRefSource {
    }
    /// Check if a file path is allowed (matches configured prefixes)
    #[allow(dead_code)]
    fn is_allowed_path(&self, path: &str) -> bool {
        self.allowed_paths.iter().any(|prefix| path.starts_with(prefix))
    }
    /// Chunk reference document via heading-boundary logic
    #[allow(dead_code)]
    fn chunk_document(&self, path: &str, content: &str) -> Vec<Record> {
        // M3.6.1 heading-boundary chunking
        // - Split by headings
@@ -203,7 +207,7 @@ mod tests {
        let chunks = source.chunk_document("docs/test.md", content);
        // Should split by headings
-        assert!(chunks.len() > 0);
+        assert!(!chunks.is_empty());
    }
    #[test]
@@ -60,8 +60,7 @@ impl MetricsCollector {
        self.by_project
            .lock()
            .unwrap()
-            .get(project)
+            .get(project).cloned()
            .map(|m| m.clone())
    }
    /// Get all project metrics.
@@ -306,7 +306,7 @@ impl QueryMetricsRepository {
        let mut repo = self.metrics.lock().unwrap();
        repo.get_mut(query_id)
            .ok_or_else(|| format!("Query {} not found", query_id))
-            .map(|metrics| f(metrics))
+            .map(f)
    }
    /// Get progress for a query
@@ -4,9 +4,10 @@
 ///
 /// Used to scope queries to project namespaces and enable graph traversal.
 /// For example: poimen/tools/kubectl.md [[debugging.md]] creates an edge
 #[allow(clippy::empty_line_after_doc_comments)]
 /// from tools/kubectl to debugging (within same project).
-use anyhow::{anyhow, Result};
+use anyhow::Result;
 use regex::Regex;
 use std::collections::{HashMap, HashSet};
 use std::path::{Path, PathBuf};
@@ -79,6 +80,7 @@ impl WikiLinkParser {
 }
 /// Graph Index: Stores and queries wiki-link relationships
 #[allow(dead_code)]
 pub struct WikiLinkGraph {
    /// Forward links: source -> [targets]
    forward_links: HashMap<String, Vec<String>>,
@@ -100,11 +102,11 @@ impl WikiLinkGraph {
    /// Add a wiki-link edge
    pub fn add_link(&mut self, source: &str, target: &str) {
        self.forward_links.entry(source.to_string())
-            .or_insert_with(Vec::new)
+            .or_default()
            .push(target.to_string());
        self.backward_links.entry(target.to_string())
-            .or_insert_with(Vec::new)
+            .or_default()
            .push(source.to_string());
    }
@@ -34,7 +34,7 @@ pub enum AuthMode {
 impl AuthMode {
    /// Detect from base URL or explicit env var.
-    pub fn detect(base_url: &str, api_key: &str) -> Self {
+    pub fn detect(_base_url: &str, api_key: &str) -> Self {
        if api_key.is_empty() {
            return Self::None;
        }
@@ -87,6 +87,7 @@ struct Choice {
 }
 #[derive(Debug, Deserialize)]
 #[allow(dead_code)]
 struct MessageResponse {
    role: String,
    content: String,
@@ -208,12 +209,11 @@ impl ChatClient {
                Ok(r) => r,
                Err(e) => {
                    last_error = Some(anyhow!("Request failed: {}", e));
-                    if e.is_timeout() || e.is_status() {
+                    if (e.is_timeout() || e.is_status())
-                        if attempt < self.max_retries - 1 {
+                        && attempt < self.max_retries - 1 {
                            tokio::time::sleep(Duration::from_millis(100 * 2_u64.pow(attempt))).await;
                            continue;
                        }
                    }
                    return Err(last_error.unwrap());
                }
            };
@@ -27,6 +27,7 @@ struct EmbeddingRequest {
 }
 #[derive(Debug, Deserialize)]
 #[allow(dead_code)]
 #[serde(untagged)]
 enum EmbeddingResponse {
    Success {
@@ -42,6 +43,7 @@ enum EmbeddingResponse {
 }
 #[derive(Debug, Deserialize)]
 #[allow(dead_code)]
 struct EmbeddingData {
    embedding: Vec<f32>,
    #[serde(default)]
@@ -120,10 +122,10 @@ impl EmbeddingsClient {
    /// Embed a single text string, returning a 768-dim vector
    pub async fn embed_one(&self, text: &str) -> Result<Vector> {
        let embeddings = self.embed(&[text.to_string()]).await?;
-        Ok(embeddings
+        embeddings
            .into_iter()
            .next()
-            .ok_or_else(|| anyhow!("empty embedding response"))?)
+            .ok_or_else(|| anyhow!("empty embedding response"))
    }
    /// Embed multiple texts, batched at ≤32 per request, preserving input order
@@ -212,4 +214,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"),
        }
    }
 }
@@ -0,0 +1,358 @@
 use anyhow::Result;
 use sqlx::{PgPool, FromRow};
 use uuid::Uuid;
 use serde::{Deserialize, Serialize};
 use chrono::{DateTime, Utc};
 #[derive(Debug, Clone, Serialize, Deserialize, FromRow)]
 pub struct AgentPrompt {
    pub id: Uuid,
    pub project_id: String,
    pub name: String,
    pub template: String,
    pub target_model: Option<String>,
    pub task_category: String,
    pub usage_count: i64,
    pub avg_quality: f32,
    pub last_used: Option<DateTime<Utc>>,
    pub active: bool,
    pub version: i32,
    pub tags: Vec<String>,
    pub created_at: DateTime<Utc>,
    pub updated_at: DateTime<Utc>,
 }
 #[derive(Debug, Clone, Serialize, Deserialize, FromRow)]
 pub struct AgentSkill {
    pub id: Uuid,
    pub project_id: String,
    pub agent_id: String,
    pub name: String,
    pub description: String,
    pub trigger_patterns: Vec<String>,
    pub success_rate: f32,
    pub invocation_count: i64,
    pub avg_latency_ms: i64,
    pub linked_prompts: Vec<Uuid>,
    pub enabled: bool,
    pub created_at: DateTime<Utc>,
    pub updated_at: DateTime<Utc>,
 }
 #[derive(Debug, Clone, Serialize, Deserialize, FromRow)]
 pub struct AgentDecision {
    pub id: Uuid,
    pub project_id: String,
    pub agent_id: String,
    pub action: String,
    pub reasoning: String,
    pub alternatives: Vec<String>,
    pub confidence: f32,
    pub context_entities: Vec<Uuid>,
    pub tool: Option<String>,
    pub task: Option<String>,
    pub outcome_success: Option<bool>,
    pub outcome_quality: Option<f32>,
    pub outcome_feedback: Option<String>,
    pub outcome_recorded_at: Option<DateTime<Utc>>,
    pub created_at: DateTime<Utc>,
    pub updated_at: DateTime<Utc>,
 }
 #[derive(Debug, Clone, Serialize, Deserialize, FromRow)]
 pub struct RolePromptMapping {
    pub id: Uuid,
    pub project_id: String,
    pub role_name: String,
    pub prompt_id: Uuid,
    pub priority: i32,
    pub active: bool,
    pub created_at: DateTime<Utc>,
    pub updated_at: DateTime<Utc>,
 }
 #[derive(Debug, Clone, Serialize, Deserialize, FromRow)]
 pub struct AgentMetrics {
    pub id: Uuid,
    pub project_id: String,
    pub agent_id: String,
    pub requests_total: i64,
    pub requests_success: i64,
    pub requests_failed: i64,
    pub average_latency_ms: f32,
    pub p95_latency_ms: f32,
    pub p99_latency_ms: f32,
    pub error_rate: f32,
    pub recorded_at: DateTime<Utc>,
 }
 pub struct AgentRepository {
    pool: PgPool,
 }
 impl AgentRepository {
    pub fn new(pool: PgPool) -> Self {
        AgentRepository { pool }
    }
    pub async fn create_prompt(&self, prompt: AgentPrompt) -> Result<AgentPrompt> {
        let result = sqlx::query_as::<_, AgentPrompt>(
            r#"
            INSERT INTO agent_prompt
            (project_id, name, template, target_model, task_category, active, version, tags)
            VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
            RETURNING *
            "#,
        )
        .bind(&prompt.project_id)
        .bind(&prompt.name)
        .bind(&prompt.template)
        .bind(&prompt.target_model)
        .bind(&prompt.task_category)
        .bind(prompt.active)
        .bind(prompt.version)
        .bind(&prompt.tags)
        .fetch_one(&self.pool)
        .await?;
        Ok(result)
    }
    pub async fn get_prompt(&self, id: Uuid) -> Result<Option<AgentPrompt>> {
        let result = sqlx::query_as::<_, AgentPrompt>(
            "SELECT * FROM agent_prompt WHERE id = $1"
        )
        .bind(id)
        .fetch_optional(&self.pool)
        .await?;
        Ok(result)
    }
    pub async fn list_prompts(&self, project_id: &str) -> Result<Vec<AgentPrompt>> {
        let results = sqlx::query_as::<_, AgentPrompt>(
            "SELECT * FROM agent_prompt WHERE project_id = $1 AND active = true ORDER BY created_at DESC"
        )
        .bind(project_id)
        .fetch_all(&self.pool)
        .await?;
        Ok(results)
    }
    pub async fn update_prompt_usage(&self, id: Uuid, quality_score: f32) -> Result<()> {
        sqlx::query(
            r#"
            UPDATE agent_prompt
            SET usage_count = usage_count + 1,
                avg_quality = (avg_quality * (usage_count) + $2) / (usage_count + 1),
                last_used = NOW(),
                updated_at = NOW()
            WHERE id = $1
            "#,
        )
        .bind(id)
        .bind(quality_score)
        .execute(&self.pool)
        .await?;
        Ok(())
    }
    pub async fn create_skill(&self, skill: AgentSkill) -> Result<AgentSkill> {
        let result = sqlx::query_as::<_, AgentSkill>(
            r#"
            INSERT INTO agent_skill
            (project_id, agent_id, name, description, enabled)
            VALUES ($1, $2, $3, $4, $5)
            RETURNING *
            "#,
        )
        .bind(&skill.project_id)
        .bind(&skill.agent_id)
        .bind(&skill.name)
        .bind(&skill.description)
        .bind(skill.enabled)
        .fetch_one(&self.pool)
        .await?;
        Ok(result)
    }
    pub async fn get_skill(&self, id: Uuid) -> Result<Option<AgentSkill>> {
        let result = sqlx::query_as::<_, AgentSkill>(
            "SELECT * FROM agent_skill WHERE id = $1"
        )
        .bind(id)
        .fetch_optional(&self.pool)
        .await?;
        Ok(result)
    }
    pub async fn list_skills(&self, project_id: &str, agent_id: &str) -> Result<Vec<AgentSkill>> {
        let results = sqlx::query_as::<_, AgentSkill>(
            "SELECT * FROM agent_skill WHERE project_id = $1 AND agent_id = $2 AND enabled = true ORDER BY created_at DESC"
        )
        .bind(project_id)
        .bind(agent_id)
        .fetch_all(&self.pool)
        .await?;
        Ok(results)
    }
    pub async fn create_decision(&self, decision: AgentDecision) -> Result<AgentDecision> {
        let result = sqlx::query_as::<_, AgentDecision>(
            r#"
            INSERT INTO agent_decision
            (project_id, agent_id, action, reasoning, confidence, tool, task)
            VALUES ($1, $2, $3, $4, $5, $6, $7)
            RETURNING *
            "#,
        )
        .bind(&decision.project_id)
        .bind(&decision.agent_id)
        .bind(&decision.action)
        .bind(&decision.reasoning)
        .bind(decision.confidence)
        .bind(&decision.tool)
        .bind(&decision.task)
        .fetch_one(&self.pool)
        .await?;
        Ok(result)
    }
    pub async fn record_decision_outcome(
        &self,
        id: Uuid,
        success: bool,
        quality: f32,
        feedback: Option<&str>,
    ) -> Result<()> {
        sqlx::query(
            r#"
            UPDATE agent_decision
            SET outcome_success = $2,
                outcome_quality = $3,
                outcome_feedback = $4,
                outcome_recorded_at = NOW(),
                updated_at = NOW()
            WHERE id = $1
            "#,
        )
        .bind(id)
        .bind(success)
        .bind(quality)
        .bind(feedback)
        .execute(&self.pool)
        .await?;
        Ok(())
    }
    pub async fn create_role_mapping(&self, mapping: RolePromptMapping) -> Result<RolePromptMapping> {
        let result = sqlx::query_as::<_, RolePromptMapping>(
            r#"
            INSERT INTO role_prompt_mapping
            (project_id, role_name, prompt_id, priority, active)
            VALUES ($1, $2, $3, $4, $5)
            RETURNING *
            "#,
        )
        .bind(&mapping.project_id)
        .bind(&mapping.role_name)
        .bind(mapping.prompt_id)
        .bind(mapping.priority)
        .bind(mapping.active)
        .fetch_one(&self.pool)
        .await?;
        Ok(result)
    }
    pub async fn get_prompts_for_role(&self, project_id: &str, role_name: &str) -> Result<Vec<AgentPrompt>> {
        let results = sqlx::query_as::<_, AgentPrompt>(
            r#"
            SELECT ap.* FROM agent_prompt ap
            INNER JOIN role_prompt_mapping rpm ON ap.id = rpm.prompt_id
            WHERE rpm.project_id = $1 AND rpm.role_name = $2 AND rpm.active = true
            ORDER BY rpm.priority DESC, ap.created_at DESC
            "#,
        )
        .bind(project_id)
        .bind(role_name)
        .fetch_all(&self.pool)
        .await?;
        Ok(results)
    }
    pub async fn save_metrics(&self, metrics: AgentMetrics) -> Result<()> {
        sqlx::query(
            r#"
            INSERT INTO agent_metrics
            (project_id, agent_id, requests_total, requests_success, requests_failed,
             average_latency_ms, p95_latency_ms, p99_latency_ms, error_rate)
            VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
            ON CONFLICT (project_id, agent_id, DATE(recorded_at)) DO UPDATE SET
            requests_total = EXCLUDED.requests_total,
            requests_success = EXCLUDED.requests_success,
            requests_failed = EXCLUDED.requests_failed,
            average_latency_ms = EXCLUDED.average_latency_ms,
            p95_latency_ms = EXCLUDED.p95_latency_ms,
            p99_latency_ms = EXCLUDED.p99_latency_ms,
            error_rate = EXCLUDED.error_rate
            "#,
        )
        .bind(&metrics.project_id)
        .bind(&metrics.agent_id)
        .bind(metrics.requests_total)
        .bind(metrics.requests_success)
        .bind(metrics.requests_failed)
        .bind(metrics.average_latency_ms)
        .bind(metrics.p95_latency_ms)
        .bind(metrics.p99_latency_ms)
        .bind(metrics.error_rate)
        .execute(&self.pool)
        .await?;
        Ok(())
    }
    pub async fn log_prompt_usage(
        &self,
        project_id: &str,
        prompt_id: Uuid,
        agent_id: Option<&str>,
        model: Option<&str>,
        input_tokens: Option<i32>,
        output_tokens: Option<i32>,
        quality_score: Option<f32>,
        duration_ms: i64,
        error_message: Option<&str>,
    ) -> Result<()> {
        sqlx::query(
            r#"
            INSERT INTO prompt_usage_log
            (project_id, prompt_id, agent_id, model_used, input_tokens, output_tokens,
             quality_score, duration_ms, error_message)
            VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
            "#,
        )
        .bind(project_id)
        .bind(prompt_id)
        .bind(agent_id)
        .bind(model)
        .bind(input_tokens)
        .bind(output_tokens)
        .bind(quality_score)
        .bind(duration_ms)
        .bind(error_message)
        .execute(&self.pool)
        .await?;
        Ok(())
    }
 }
@@ -1,7 +1,6 @@
 use chrono::{DateTime, Utc};
 use sqlx::PgPool;
 use uuid::Uuid;
 use serde_json::json;
 /// Minimal audit logger - records version snapshots on mutation
 #[derive(Clone)]
@@ -8,6 +8,7 @@ pub mod edge_repo;
 pub mod community_repo;
 pub mod versioning;
 pub mod audit_logger;
 pub mod agent_repo;
 // pub mod db_repo;  // TODO: Fix Entity schema integration
 pub use event_log::{EventRecord, LogWriter};
@@ -0,0 +1,21 @@
 version: '3.8'
 services:
  postgres-test:
    image: postgres:16-alpine
    environment:
      POSTGRES_USER: app
      POSTGRES_PASSWORD: testpass
      POSTGRES_DB: memory
    ports:
      - "5433:5432"
    volumes:
      - postgres-test-data:/var/lib/postgresql/data
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U app -d memory"]
      interval: 2s
      timeout: 5s
      retries: 10
 volumes:
  postgres-test-data:
@@ -0,0 +1,131 @@
 apiVersion: tekton.dev/v1
 kind: Task
 metadata:
  name: agent-memory-migration
  namespace: tekton-pipelines
 spec:
  description: Apply agent memory schema migration (004) to production database
  params:
    - name: migration-version
      description: Migration version number
      default: "004"
    - name: database-name
      description: Database name
      default: "memory"
  workspaces:
    - name: source
      description: Git source with migrations
    - name: db-credentials
      description: Database credentials secret
  steps:
    - name: apply-migration
      image: postgres:16-alpine
      workingDir: $(workspaces.source.path)
      env:
        - name: PGPASSWORD
          valueFrom:
            secretKeyRef:
              name: memory-db-app
              key: password
        - name: PGHOST
          value: memory-db-rw.poimen.svc.cluster.local
        - name: PGUSER
          value: app
        - name: PGDATABASE
          value: $(params.database-name)
      script: |
        #!/bin/sh
        set -e
        echo "Applying migration $(params.migration-version)_agent_memory_schema.sql"
        # Wait for database to be ready
        until pg_isready -h $PGHOST -U $PGUSER -d $PGDATABASE; do
          echo "Waiting for database..."
          sleep 2
        done
        # Apply migration
        psql -h $PGHOST -U $PGUSER -d $PGDATABASE \
          -f migrations/$(params.migration-version)_agent_memory_schema.sql
        # Verify tables created
        TABLES=$(psql -h $PGHOST -U $PGUSER -d $PGDATABASE -t -c \
          "SELECT count(*) FROM information_schema.tables WHERE table_schema='public' AND table_name IN ('agent_prompt', 'agent_skill', 'agent_decision', 'role_prompt_mapping', 'agent_metrics')")
        if [ "$TABLES" -eq 5 ]; then
          echo "✓ All agent memory tables created successfully"
          exit 0
        else
          echo "✗ Migration failed: expected 5 tables, found $TABLES"
          exit 1
        fi
    - name: verify-indexes
      image: postgres:16-alpine
      env:
        - name: PGPASSWORD
          valueFrom:
            secretKeyRef:
              name: memory-db-app
              key: password
        - name: PGHOST
          value: memory-db-rw.poimen.svc.cluster.local
        - name: PGUSER
          value: app
        - name: PGDATABASE
          value: $(params.database-name)
      script: |
        #!/bin/sh
        set -e
        echo "Verifying indexes..."
        INDEXES=$(psql -h $PGHOST -U $PGUSER -d $PGDATABASE -t -c \
          "SELECT count(*) FROM pg_indexes WHERE schemaname='public' AND tablename LIKE 'agent_%'")
        if [ "$INDEXES" -gt 0 ]; then
          echo "✓ Found $INDEXES indexes on agent tables"
          psql -h $PGHOST -U $PGUSER -d $PGDATABASE -c \
            "SELECT indexname FROM pg_indexes WHERE schemaname='public' AND tablename LIKE 'agent_%' ORDER BY indexname;" 
        else
          echo "✗ No indexes found on agent tables"
          exit 1
        fi
    - name: verify-schemas
      image: postgres:16-alpine
      env:
        - name: PGPASSWORD
          valueFrom:
            secretKeyRef:
              name: memory-db-app
              key: password
        - name: PGHOST
          value: memory-db-rw.poimen.svc.cluster.local
        - name: PGUSER
          value: app
        - name: PGDATABASE
          value: $(params.database-name)
      script: |
        #!/bin/sh
        set -e
        echo "Verifying table schemas..."
        # Verify agent_prompt table
        psql -h $PGHOST -U $PGUSER -d $PGDATABASE -c "
          SELECT column_name, data_type, is_nullable
          FROM information_schema.columns
          WHERE table_name='agent_prompt'
          ORDER BY ordinal_position;"
        echo "✓ Agent prompt schema verified"
        # Verify role_prompt_mapping has foreign key
        psql -h $PGHOST -U $PGUSER -d $PGDATABASE -c "
          SELECT constraint_name, constraint_type
          FROM information_schema.table_constraints
          WHERE table_name='role_prompt_mapping';"
        echo "✓ All table schemas verified"
@@ -0,0 +1,76 @@
 ---
 # PipelineRun: Agent Memory Feature Testing
 # Tests role-to-prompt mapping with API Platform Engineer role requirements
 # Runs migrations, integration tests, and validates all constraints
 apiVersion: tekton.dev/v1
 kind: PipelineRun
 metadata:
  name: agent-memory-test-run
  namespace: poimen
  generateName: agent-memory-test-
 spec:
  pipelineRef:
    name: poimen-ci
  params:
  - name: image
    value: "forgejo.riotpiao.com/riotpiao-poimen/poimen-memory:latest"
  - name: registry-user
    value: "riotpiao-poimen"
  - name: registry-token
    value: "${FORGEJO_REGISTRY_TOKEN}"  # Injected by ArgoCD/SOPS
  workspaces:
  - name: source
    emptyDir: {}  # Or use PVC for persistent builds
  serviceAccountName: tekton-builder
  timeouts:
    pipeline: "1h"
    tasks: "30m"
 ---
 # ServiceAccount for Tekton Pipeline (builder with DB access)
 apiVersion: v1
 kind: ServiceAccount
 metadata:
  name: tekton-builder
  namespace: poimen
 ---
 # ClusterRoleBinding: Allow pipeline to query database via pod exec
 apiVersion: rbac.authorization.k8s.io/v1
 kind: ClusterRoleBinding
 metadata:
  name: tekton-builder-db-access
 roleRef:
  apiGroup: rbac.authorization.k8s.io
  kind: ClusterRole
  name: tekton-builder-db-access
 subjects:
 - kind: ServiceAccount
  name: tekton-builder
  namespace: poimen
 ---
 # ClusterRole: Database access for migrations
 apiVersion: rbac.authorization.k8s.io/v1
 kind: ClusterRole
 metadata:
  name: tekton-builder-db-access
 rules:
 - apiGroups: [""]
  resources: ["pods"]
  verbs: ["get", "list"]
 - apiGroups: [""]
  resources: ["pods/exec"]
  verbs: ["create"]
 - apiGroups: [""]
  resources: ["secrets"]
  resourceNames: ["memory-db-app"]
  verbs: ["get"]
 - apiGroups: [""]
  resources: ["services"]
  verbs: ["get", "list"]
@@ -0,0 +1,144 @@
 ---
 # 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:
@@ -0,0 +1,140 @@
 ---
 # 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: ""
  workspaces:
  - name: source
    description: "Git source repository with migrations"
  tasks:
  # Task 0: Apply Agent Memory Migrations
  - name: agent-memory-migration
    taskRef:
      name: agent-memory-migration
    params:
    - name: migration-version
      value: "004"
    - name: database-name
      value: "memory"
    workspaces:
    - name: source
      workspace: source
  # Task 1: Integration Tests (runs after migration)
  - name: integration-tests
    runAfter:
      - agent-memory-migration
    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"
@@ -0,0 +1,144 @@
 -- Agent Memory Schema (Phase 6)
 -- Stores agent prompts, skills, and decisions with role-to-prompt mapping
 -- Follows API Platform Engineer contract-first design (agency-agents role)
 CREATE TABLE IF NOT EXISTS agent_prompt (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  project_id VARCHAR(255) NOT NULL,
  name VARCHAR(512) NOT NULL,
  template TEXT NOT NULL,
  target_model VARCHAR(128),
  task_category VARCHAR(128) NOT NULL,
  usage_count BIGINT DEFAULT 0,
  avg_quality FLOAT DEFAULT 0.0,
  last_used TIMESTAMP WITH TIME ZONE,
  active BOOLEAN DEFAULT true,
  version INTEGER DEFAULT 1,
  tags TEXT[] DEFAULT '{}',
  created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(),
  updated_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(),
  UNIQUE(project_id, name, version)
 );
 CREATE INDEX idx_agent_prompt_project_active ON agent_prompt(project_id, active);
 CREATE INDEX idx_agent_prompt_task_category ON agent_prompt(task_category);
 CREATE INDEX idx_agent_prompt_tags ON agent_prompt USING GIN(tags);
 -- Agent Skill: linked capabilities with effectiveness tracking
 CREATE TABLE IF NOT EXISTS agent_skill (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  project_id VARCHAR(255) NOT NULL,
  agent_id VARCHAR(255) NOT NULL,
  name VARCHAR(512) NOT NULL,
  description TEXT NOT NULL,
  trigger_patterns TEXT[] DEFAULT '{}',
  success_rate FLOAT DEFAULT 0.0,
  invocation_count BIGINT DEFAULT 0,
  avg_latency_ms BIGINT DEFAULT 0,
  linked_prompts UUID[] DEFAULT '{}',
  enabled BOOLEAN DEFAULT true,
  created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(),
  updated_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(),
  UNIQUE(project_id, agent_id, name)
 );
 CREATE INDEX idx_agent_skill_agent ON agent_skill(project_id, agent_id);
 CREATE INDEX idx_agent_skill_enabled ON agent_skill(enabled);
 CREATE INDEX idx_agent_skill_linked_prompts ON agent_skill USING GIN(linked_prompts);
 -- Agent Decision: reasoning and outcome tracking
 CREATE TABLE IF NOT EXISTS agent_decision (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  project_id VARCHAR(255) NOT NULL,
  agent_id VARCHAR(255) NOT NULL,
  action VARCHAR(512) NOT NULL,
  reasoning TEXT NOT NULL,
  alternatives TEXT[] DEFAULT '{}',
  confidence FLOAT DEFAULT 0.0,
  context_entities UUID[] DEFAULT '{}',
  tool VARCHAR(255),
  task VARCHAR(255),
  outcome_success BOOLEAN,
  outcome_quality FLOAT,
  outcome_feedback TEXT,
  outcome_recorded_at TIMESTAMP WITH TIME ZONE,
  created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(),
  updated_at TIMESTAMP WITH TIME ZONE DEFAULT NOW()
 );
 CREATE INDEX idx_agent_decision_agent ON agent_decision(project_id, agent_id);
 CREATE INDEX idx_agent_decision_action ON agent_decision(action);
 CREATE INDEX idx_agent_decision_context ON agent_decision USING GIN(context_entities);
 -- Agent Registration: lifecycle management
 CREATE TABLE IF NOT EXISTS agent_registry (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  project_id VARCHAR(255) NOT NULL,
  agent_id VARCHAR(255) NOT NULL,
  capabilities TEXT[] NOT NULL,
  webhook_url VARCHAR(2048),
  rate_limit INTEGER DEFAULT 1000,
  metadata JSONB DEFAULT '{}',
  status VARCHAR(32) DEFAULT 'active',
  created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(),
  updated_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(),
  UNIQUE(project_id, agent_id)
 );
 CREATE INDEX idx_agent_registry_project ON agent_registry(project_id);
 CREATE INDEX idx_agent_registry_status ON agent_registry(status);
 -- Role-to-Prompt Mapping: maps agent roles to prompt templates
 CREATE TABLE IF NOT EXISTS role_prompt_mapping (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  project_id VARCHAR(255) NOT NULL,
  role_name VARCHAR(255) NOT NULL,
  prompt_id UUID NOT NULL REFERENCES agent_prompt(id) ON DELETE CASCADE,
  priority INTEGER DEFAULT 0,
  active BOOLEAN DEFAULT true,
  created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(),
  updated_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(),
  UNIQUE(project_id, role_name, prompt_id)
 );
 CREATE INDEX idx_role_prompt_mapping_role ON role_prompt_mapping(project_id, role_name, active);
 CREATE INDEX idx_role_prompt_mapping_prompt ON role_prompt_mapping(prompt_id);
 -- Agent Metrics: performance tracking
 CREATE TABLE IF NOT EXISTS agent_metrics (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  project_id VARCHAR(255) NOT NULL,
  agent_id VARCHAR(255) NOT NULL,
  requests_total BIGINT DEFAULT 0,
  requests_success BIGINT DEFAULT 0,
  requests_failed BIGINT DEFAULT 0,
  average_latency_ms FLOAT DEFAULT 0.0,
  p95_latency_ms FLOAT DEFAULT 0.0,
  p99_latency_ms FLOAT DEFAULT 0.0,
  error_rate FLOAT DEFAULT 0.0,
  recorded_at TIMESTAMP WITH TIME ZONE DEFAULT NOW()
 );
 -- Note: Use daily rollup job or materialized view for DATE(recorded_at) unique constraint
 -- PostgreSQL doesn't allow functions in UNIQUE constraints, so we use trigger-based rollup instead
 CREATE INDEX idx_agent_metrics_agent ON agent_metrics(project_id, agent_id, recorded_at DESC);
 -- Prompt Usage Log: detailed invocation tracking
 CREATE TABLE IF NOT EXISTS prompt_usage_log (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  project_id VARCHAR(255) NOT NULL,
  prompt_id UUID NOT NULL REFERENCES agent_prompt(id) ON DELETE CASCADE,
  agent_id VARCHAR(255),
  model_used VARCHAR(128),
  input_tokens INTEGER,
  output_tokens INTEGER,
  quality_score FLOAT,
  duration_ms BIGINT,
  error_message TEXT,
  created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW()
 );
 CREATE INDEX idx_prompt_usage_log_prompt ON prompt_usage_log(prompt_id, created_at DESC);
 CREATE INDEX idx_prompt_usage_log_agent ON prompt_usage_log(agent_id, created_at DESC);
 CREATE INDEX idx_prompt_usage_log_project ON prompt_usage_log(project_id, created_at DESC);
@@ -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
@@ -0,0 +1,608 @@
 // Integration test: Agent Memory with API Platform Engineer role requirements
 // Tests contract-first design per agency-agents/engineering/engineering-api-platform-engineer.md
 #[cfg(test)]
 mod tests {
    use serde_json::{json, Value};
    // Test constants aligned with API Platform Engineer role
    const API_VERSION: &str = "v1";
    const PROJECT_ID: &str = "poimen";
    const TEST_AGENT_ID: &str = "api-platform-engineer";
    const API_PLATFORM_ENGINEER_ROLE: &str = "api-platform-engineer";
    // API Platform Engineer role prompt templates
    const CONTRACT_FIRST_PROMPT: &str = r#"
 You are an API Platform Engineer designing a contract-first API.
 Task: Review the following API specification for:
 1. Naming consistency (pick snake_case or camelCase and never waver)
 2. Backward compatibility (no breaking changes without versioning)
 3. Error responses (consistent structure, stable codes, correct HTTP status semantics)
 4. Rate limiting (communicated, not just enforced)
 5. Documentation (SDKs and docs generated from spec, never drift)
 Specification:
 {{spec}}
 Output JSON with:
 {
  "contract_valid": boolean,
  "breaking_changes": [string],
  "naming_inconsistencies": [string],
  "error_issues": [string],
  "rate_limit_issues": [string],
  "recommendations": [string]
 }
 "#;
    const BACKWARD_COMPATIBILITY_PROMPT: &str = r#"
 You are an API versioning expert.
 Analyze the proposed change:
 {{change}}
 Determine:
 1. Is this a breaking change?
 2. Does it require a new version?
 3. What's the migration path?
 4. What deprecation runway is needed?
 Output JSON with:
 {
  "breaking": boolean,
  "requires_new_version": boolean,
  "migration_path": string,
  "deprecation_runway_days": number,
  "is_safe_additive": boolean
 }
 "#;
    const SDK_GENERATION_PROMPT: &str = r#"
 You are an SDK generation specialist.
 Given this OpenAPI spec:
 {{spec}}
 Generate SDK requirements for:
 1. Language: {{language}}
 2. Idiomatic patterns for that language
 3. Error handling
 4. Retry logic and idempotency
 5. Type safety
 Output JSON with:
 {
  "sdk_structure": object,
  "error_handling": string,
  "idempotency_strategy": string,
  "type_safety_level": string,
  "generated_package_version": string
 }
 "#;
    #[test]
    fn test_contract_first_api_specification() {
        // Contract-first principle: OpenAPI spec is source of truth
        let api_spec = json!({
            "openapi": "3.0.0",
            "info": {
                "title": "Poimen Agent Memory API",
                "version": API_VERSION,
                "description": "Agent memory with role-to-prompt mapping"
            },
            "paths": {
                "/memory/agents/{project_id}/prompts": {
                    "post": {
                        "operationId": "createPrompt",
                        "parameters": [
                            {
                                "name": "project_id",
                                "in": "path",
                                "required": true,
                                "schema": { "type": "string" }
                            }
                        ],
                        "requestBody": {
                            "required": true,
                            "content": {
                                "application/json": {
                                    "schema": {
                                        "type": "object",
                                        "required": ["name", "template", "task_category"],
                                        "properties": {
                                            "name": { "type": "string", "minLength": 1 },
                                            "template": { "type": "string", "description": "Prompt template with {{placeholders}}" },
                                            "target_model": { "type": "string", "example": "ornith:35b" },
                                            "task_category": { "type": "string", "enum": ["extraction", "reasoning", "summarization", "validation"] },
                                            "tags": { "type": "array", "items": { "type": "string" } }
                                        }
                                    }
                                }
                            }
                        },
                        "responses": {
                            "201": {
                                "description": "Prompt created",
                                "content": {
                                    "application/json": {
                                        "schema": { "$ref": "#/components/schemas/Prompt" }
                                    }
                                }
                            },
                            "400": { "$ref": "#/components/responses/BadRequest" },
                            "429": { "$ref": "#/components/responses/RateLimited" }
                        }
                    }
                },
                "/memory/agents/{project_id}/roles": {
                    "post": {
                        "operationId": "mapRoleToPrompt",
                        "requestBody": {
                            "required": true,
                            "content": {
                                "application/json": {
                                    "schema": {
                                        "type": "object",
                                        "required": ["role_name", "prompt_id"],
                                        "properties": {
                                            "role_name": { "type": "string", "minLength": 1 },
                                            "prompt_id": { "type": "string", "format": "uuid" },
                                            "priority": { "type": "integer", "default": 0 }
                                        }
                                    }
                                }
                            }
                        },
                        "responses": {
                            "200": { "description": "Mapping created" },
                            "400": { "$ref": "#/components/responses/BadRequest" }
                        }
                    }
                },
                "/memory/agents/{project_id}/roles/{role_name}/prompts": {
                    "get": {
                        "operationId": "getRolePrompts",
                        "responses": {
                            "200": { "description": "List of prompts for role" },
                            "404": { "$ref": "#/components/responses/NotFound" }
                        }
                    }
                }
            },
            "components": {
                "schemas": {
                    "Prompt": {
                        "type": "object",
                        "required": ["id", "name", "template", "task_category"],
                        "properties": {
                            "id": { "type": "string", "format": "uuid" },
                            "name": { "type": "string" },
                            "template": { "type": "string" },
                            "target_model": { "type": "string", "nullable": true },
                            "task_category": { "type": "string" },
                            "usage_count": { "type": "integer" },
                            "avg_quality": { "type": "number", "format": "float" },
                            "version": { "type": "integer" },
                            "created_at": { "type": "string", "format": "date-time" }
                        }
                    },
                    "Error": {
                        "type": "object",
                        "required": ["code", "message"],
                        "properties": {
                            "code": { "type": "string", "description": "Machine-readable error code" },
                            "message": { "type": "string", "description": "Human-readable error message" },
                            "details": { "type": "object", "description": "Field-level or contextual detail" },
                            "request_id": { "type": "string", "description": "Trace this to support" }
                        }
                    }
                },
                "responses": {
                    "BadRequest": {
                        "description": "Bad request",
                        "content": {
                            "application/json": { "schema": { "$ref": "#/components/schemas/Error" } }
                        }
                    },
                    "NotFound": {
                        "description": "Resource not found",
                        "content": {
                            "application/json": { "schema": { "$ref": "#/components/schemas/Error" } }
                        }
                    },
                    "RateLimited": {
                        "description": "Rate limited",
                        "headers": {
                            "Retry-After": { "schema": { "type": "integer" } },
                            "X-RateLimit-Limit": { "schema": { "type": "integer" } },
                            "X-RateLimit-Remaining": { "schema": { "type": "integer" } },
                            "X-RateLimit-Reset": { "schema": { "type": "integer" } }
                        },
                        "content": {
                            "application/json": { "schema": { "$ref": "#/components/schemas/Error" } }
                        }
                    }
                }
            }
        });
        // Validate contract structure
        assert_eq!(api_spec["openapi"], "3.0.0");
        assert_eq!(api_spec["info"]["version"], API_VERSION);
        // Validate error schema is consistent
        let error_schema = &api_spec["components"]["schemas"]["Error"];
        assert!(error_schema["required"]
            .as_array()
            .unwrap()
            .contains(&Value::String("code".to_string())));
        assert!(error_schema["required"]
            .as_array()
            .unwrap()
            .contains(&Value::String("message".to_string())));
        // Validate naming consistency (snake_case)
        assert!(
            api_spec["paths"]["/memory/agents/{project_id}/prompts"]["post"]["operationId"]
                .as_str()
                .unwrap()
                .contains("createPrompt")
        );
        assert!(
            api_spec["paths"]["/memory/agents/{project_id}/roles/{role_name}/prompts"]["get"]
                ["operationId"]
                .as_str()
                .unwrap()
                .contains("getRolePrompts")
        );
        // Validate backward compatibility: all fields are optional except required ones
        let create_prompt_schema = &api_spec["paths"]["/memory/agents/{project_id}/prompts"]
            ["post"]["requestBody"]["content"]["application/json"]["schema"];
        assert_eq!(
            create_prompt_schema["required"].as_array().unwrap(),
            &vec![
                Value::String("name".to_string()),
                Value::String("template".to_string()),
                Value::String("task_category".to_string())
            ]
        );
        println!("✓ Contract-first API specification validated");
    }
    #[test]
    fn test_backward_compatibility_rules() {
        // Rule 1: Adding optional fields is safe
        let safe_change = json!({
            "type": "add_field",
            "field": "metadata",
            "required": false,
            "breaking": false
        });
        assert!(!safe_change["breaking"].as_bool().unwrap());
        // Rule 2: Removing fields is breaking
        let breaking_change = json!({
            "type": "remove_field",
            "field": "template",
            "breaking": true,
            "requires_version_bump": true
        });
        assert!(breaking_change["breaking"].as_bool().unwrap());
        assert!(breaking_change["requires_version_bump"].as_bool().unwrap());
        // Rule 3: Adding new enum value is safe if clients tolerate unknowns
        let safe_enum_addition = json!({
            "type": "add_enum_value",
            "enum": "task_category",
            "new_value": "planning",
            "breaking": false,
            "requires_documentation": true
        });
        assert!(!safe_enum_addition["breaking"].as_bool().unwrap());
        // Rule 4: Changing field type is breaking
        let breaking_type_change = json!({
            "type": "change_field_type",
            "field": "usage_count",
            "old_type": "integer",
            "new_type": "string",
            "breaking": true,
            "requires_version_bump": true,
            "migration_path": "Convert all consumers to parse as string"
        });
        assert!(breaking_type_change["breaking"].as_bool().unwrap());
        println!("✓ Backward compatibility rules validated");
    }
    #[test]
    fn test_rate_limiting_communication() {
        // Rate limits must be communicated in response headers
        let response_headers = json!({
            "X-RateLimit-Limit": 1000,
            "X-RateLimit-Remaining": 847,
            "X-RateLimit-Reset": 1720483200,
            "Retry-After": 30
        });
        // All required rate limit headers present
        assert!(response_headers.get("X-RateLimit-Limit").is_some());
        assert!(response_headers.get("X-RateLimit-Remaining").is_some());
        assert!(response_headers.get("X-RateLimit-Reset").is_some());
        // On 429, Retry-After present
        let rate_limited_response = json!({
            "status": 429,
            "error": {
                "code": "rate_limit_exceeded",
                "message": "1000 req/hr exceeded; retry after 30s",
                "request_id": "req_a1b2"
            },
            "headers": {
                "Retry-After": 30
            }
        });
        assert_eq!(rate_limited_response["status"], 429);
        assert_eq!(
            rate_limited_response["error"]["code"],
            "rate_limit_exceeded"
        );
        assert!(
            rate_limited_response["headers"]["Retry-After"]
                .as_i64()
                .unwrap()
                > 0
        );
        println!("✓ Rate limiting communication validated");
    }
    #[test]
    fn test_error_response_consistency() {
        // Error responses must have consistent structure everywhere
        let errors = vec![
            json!({
                "code": "invalid_request",
                "message": "name field required",
                "details": { "field": "name" },
                "request_id": "req-123"
            }),
            json!({
                "code": "not_found",
                "message": "Prompt not found",
                "details": { "prompt_id": "uuid-456" },
                "request_id": "req-789"
            }),
            json!({
                "code": "permission_denied",
                "message": "Insufficient capabilities",
                "details": { "required": "memory:write" },
                "request_id": "req-999"
            }),
        ];
        for error in errors {
            // All errors have required structure
            assert!(error["code"].is_string());
            assert!(error["message"].is_string());
            assert!(error["request_id"].is_string());
            // No 200 with error (must use proper HTTP status)
            assert_ne!(error["code"], ""); // code is stable, machine-readable
        }
        println!("✓ Error response consistency validated");
    }
    #[test]
    fn test_deprecation_lifecycle() {
        // Deprecation requires: Announce → Signal → Runway → Monitor → Sunset
        let deprecation_plan = json!({
            "endpoint": "/agents/{id}",
            "lifecycle": {
                "phase": "announced",
                "deprecation_date": "2025-06-01",
                "sunset_date": "2026-06-01",
                "runway_days": 365
            },
            "signals": {
                "deprecation_header": "Deprecation: true",
                "sunset_header": "Sunset: Sun, 01 Jun 2026 00:00:00 GMT",
                "warning_in_response": true
            },
            "migration_guide": "Use /agents/v2/{id} instead",
            "monitoring": {
                "track_usage_by_consumer": true,
                "alert_on_remaining_usage": true
            }
        });
        assert_eq!(deprecation_plan["lifecycle"]["runway_days"], 365);
        assert!(deprecation_plan["signals"]["deprecation_header"]
            .as_str()
            .unwrap()
            .contains("Deprecation"));
        assert!(deprecation_plan["monitoring"]["track_usage_by_consumer"]
            .as_bool()
            .unwrap());
        println!("✓ Deprecation lifecycle validated");
    }
    #[test]
    fn test_idempotency_and_retry_safety() {
        // Write operations must be idempotent via Idempotency-Key
        let request_with_key = json!({
            "method": "POST",
            "path": "/memory/agents/project1/prompts",
            "headers": {
                "Idempotency-Key": "req-unique-uuid-123"
            },
            "body": {
                "name": "extract-entities",
                "template": "Extract entities from {{text}}"
            }
        });
        assert!(request_with_key["headers"]["Idempotency-Key"].is_string());
        // Retry with same key returns cached response
        let response_1 = json!({
            "status": 201,
            "id": "prompt-uuid-456"
        });
        let response_2_retry = json!({
            "status": 201,
            "id": "prompt-uuid-456",
            "cached": true
        });
        // Both return same result → safe to retry
        assert_eq!(response_1["id"], response_2_retry["id"]);
        println!("✓ Idempotency and retry safety validated");
    }
    #[test]
    fn test_api_platform_engineer_role_requirements() {
        // Comprehensive validation per api-platform-engineer.md role
        let role_requirements = json!({
            "role": API_PLATFORM_ENGINEER_ROLE,
            "requirements": {
                "contract_first": {
                    "openapi_spec": "required",
                    "source_of_truth_before_code": true,
                    "consistency_reviewed": true
                },
                "backward_compatibility": {
                    "no_silent_breaking_changes": true,
                    "additive_changes_allowed": true,
                    "versioning_policy": "major version in path (/v1, /v2)",
                    "deprecation_runway": "6-12+ months"
                },
                "error_handling": {
                    "consistent_structure": true,
                    "stable_machine_readable_code": true,
                    "correct_http_status": true,
                    "request_id_for_tracing": true
                },
                "rate_limiting": {
                    "communicated_headers": true,
                    "no_ambush_429": true,
                    "retry_after_provided": true
                },
                "sdk_and_docs": {
                    "generated_from_spec": true,
                    "never_drift": true,
                    "typed_idiomatic": true,
                    "multiple_languages": true
                },
                "idempotency": {
                    "write_operations_idempotent": true,
                    "idempotency_key_support": true,
                    "safe_retry": true
                }
            }
        });
        // Validate all requirements
        assert!(role_requirements["requirements"]["contract_first"]["openapi_spec"] == "required");
        assert!(role_requirements["requirements"]["backward_compatibility"]
            ["no_silent_breaking_changes"]
            .as_bool()
            .unwrap());
        assert!(
            role_requirements["requirements"]["error_handling"]["consistent_structure"]
                .as_bool()
                .unwrap()
        );
        assert!(
            role_requirements["requirements"]["rate_limiting"]["communicated_headers"]
                .as_bool()
                .unwrap()
        );
        assert!(
            role_requirements["requirements"]["sdk_and_docs"]["generated_from_spec"]
                .as_bool()
                .unwrap()
        );
        assert!(
            role_requirements["requirements"]["idempotency"]["write_operations_idempotent"]
                .as_bool()
                .unwrap()
        );
        println!("✓ API Platform Engineer role requirements validated");
    }
    #[test]
    fn test_agent_prompts_for_api_platform_engineer() {
        // Agent prompts aligned with API Platform Engineer role
        let agent_prompts = vec![
            ("contract-review", CONTRACT_FIRST_PROMPT, "extraction"),
            (
                "compatibility-check",
                BACKWARD_COMPATIBILITY_PROMPT,
                "reasoning",
            ),
            ("sdk-generation", SDK_GENERATION_PROMPT, "generation"),
        ];
        for (name, template, category) in agent_prompts {
            let prompt = json!({
                "name": name,
                "template": template,
                "task_category": category,
                "target_model": "ornith:35b"
            });
            assert!(!prompt["template"].as_str().unwrap().is_empty());
            assert!(
                prompt["template"].as_str().unwrap().contains("{{")
                    || prompt["template"].as_str().unwrap().contains("output")
            );
        }
        println!("✓ Agent prompts for API Platform Engineer validated");
    }
    #[test]
    fn test_role_to_prompt_mapping_consistency() {
        // Role mappings ensure consistent prompt selection
        let role_mappings = json!({
            "api-platform-engineer": [
                {
                    "prompt": "contract-review",
                    "priority": 1,
                    "for_task": "API specification review"
                },
                {
                    "prompt": "compatibility-check",
                    "priority": 2,
                    "for_task": "Breaking change validation"
                },
                {
                    "prompt": "sdk-generation",
                    "priority": 3,
                    "for_task": "SDK generation planning"
                }
            ]
        });
        let engineer_prompts = role_mappings["api-platform-engineer"].as_array().unwrap();
        assert_eq!(engineer_prompts.len(), 3);
        // Prompts ordered by priority
        assert!(
            engineer_prompts[0]["priority"].as_i64().unwrap()
                < engineer_prompts[1]["priority"].as_i64().unwrap()
        );
        println!("✓ Role-to-prompt mapping consistency validated");
    }
 }