Compare commits

..
Author SHA1 Message Date
rock f34e96d171 test: verify embedding response parsing against real service format
CI / CI (pull_request) Successful in 12m48s
- 6 parsing tests for EmbeddingResponse struct
- test_parse_real_embedding_response: exact format from embeddings-predictor
- test_parse_768_dim_response: full 768-dim vector
- test_parse_multi_input_response: array input returns multiple embeddings
- test_parse_embedding_error_response: error format
- test_parse_html_fails_gracefully: HTML error page correctly rejected
- Confirms: parsing is correct, 'expected ident' error is non-JSON response
2026-09-14 08:46:03 +09:00
49 changed files with 785 additions and 2518 deletions
+1 -1
View File
@@ -4,7 +4,7 @@ MEM_RATE_LIMIT_QUERY=10000
MEM_IDEMPOTENCY_TTL_SECS=86400 MEM_IDEMPOTENCY_TTL_SECS=86400
MEM_EMBEDDING_BATCH_SIZE=4 MEM_EMBEDDING_BATCH_SIZE=4
DATABASE_URL=postgresql://app:***REMOVED***@127.0.0.1:5433/memory DATABASE_URL=postgresql://app:katFpWYB4EH9KU9NABOglnE9ekea5rBxyOY9WZeUTi1ujhFS1pVzNxrXbB7A4qGc@127.0.0.1:5433/memory
# Embedding via direct port-forward (skip gateway auth) # Embedding via direct port-forward (skip gateway auth)
LLM_ENDPOINT=http://localhost:9090/v1/chat/completions LLM_ENDPOINT=http://localhost:9090/v1/chat/completions
+1 -105
View File
@@ -71,111 +71,7 @@ jobs:
docker push "${IMAGE}:${{ steps.sha.outputs.short_sha }}" docker push "${IMAGE}:${{ steps.sha.outputs.short_sha }}"
echo "Pushed: ${IMAGE}:${{ steps.sha.outputs.short_sha }}" echo "Pushed: ${IMAGE}:${{ steps.sha.outputs.short_sha }}"
- name: Install kubectl - name: Prune unused images and cleanup
run: |
apt-get update
apt-get install -y kubectl
- name: Setup kubeconfig for Tekton
run: |
mkdir -p ~/.kube
echo "${KUBECONFIG_B64}" | base64 -d > ~/.kube/config
chmod 600 ~/.kube/config
kubectl cluster-info 2>&1 | head -3
echo "✓ kubeconfig ready"
env:
KUBECONFIG_B64: ${{ secrets.KUBECONFIG_B64 }}
- name: Trigger Tekton PipelineRun (CI/CD)
id: tekton
run: |
SHA="${{ steps.sha.outputs.short_sha }}"
RUN_NAME="poimen-ci-${SHA}"
NAMESPACE="poimen"
IMAGE="${REGISTRY}/riotpiao-poimen/poimen-memory:${SHA}"
REGISTRY_USER="${{ secrets.FORGEJO_REGISTRY_USER }}"
REGISTRY_TOKEN="${{ secrets.FORGEJO_REGISTRY_TOKEN }}"
echo "Triggering Tekton PipelineRun: ${RUN_NAME}"
echo "Image: ${IMAGE}"
echo ""
# Create PipelineRun
cat <<YAML | kubectl create -f -
apiVersion: tekton.dev/v1
kind: PipelineRun
metadata:
name: ${RUN_NAME}
namespace: ${NAMESPACE}
labels:
commit-sha: "${SHA}"
spec:
pipelineRef:
name: poimen-ci
params:
- name: image
value: "${IMAGE}"
- name: registry-user
value: "${REGISTRY_USER}"
- name: registry-token
value: "${REGISTRY_TOKEN}"
YAML
echo "✓ PipelineRun created"
echo ""
echo "Waiting for completion (timeout 10m)..."
# Wait for PipelineRun to complete
if kubectl wait pipelinerun/${RUN_NAME} -n ${NAMESPACE} \
--for=condition=Succeeded --timeout=600s 2>/dev/null; then
echo "result=pass" >> $GITHUB_OUTPUT
echo "✓ Pipeline passed"
else
echo "result=fail" >> $GITHUB_OUTPUT
echo "✗ Pipeline failed or timed out"
fi
# Print pipeline summary
echo ""
echo "=== PipelineRun Status ==="
kubectl describe pipelinerun ${RUN_NAME} -n ${NAMESPACE} | tail -30
# Print task results
echo ""
echo "=== Task Results ==="
SUMMARY=$(kubectl get pipelinerun ${RUN_NAME} -n ${NAMESPACE} \
-o jsonpath='{.status.taskRuns[*].status.taskResults[?(@.name=="summary")].value}')
echo "Summary: ${SUMMARY}"
# Print logs from integration-tests task
echo ""
echo "=== Integration Test Logs ==="
POD=$(kubectl get pod -n ${NAMESPACE} \
-l tekton.dev/pipelineRun=${RUN_NAME} -l tekton.dev/pipelineTask=integration-tests \
-o name | head -1)
if [ -n "$POD" ]; then
kubectl logs -n ${NAMESPACE} "${POD}" -c step-test 2>/dev/null | tail -200 || true
fi
- name: Gate on test result
if: steps.tekton.outputs.result != 'pass'
run: |
echo "✗ Integration tests FAILED"
echo "Image NOT promoted to :latest"
exit 1
- name: Promote image to latest
run: |
docker login -u "${REGISTRY_USER}" -p "${REGISTRY_TOKEN}" "${REGISTRY}"
docker tag "${IMAGE}:${{ steps.sha.outputs.short_sha }}" "${IMAGE}:latest"
docker push "${IMAGE}:latest"
echo "✓ Promoted to :latest"
env:
REGISTRY_USER: ${{ secrets.FORGEJO_REGISTRY_USER }}
REGISTRY_TOKEN: ${{ secrets.FORGEJO_REGISTRY_TOKEN }}
- name: Cleanup
if: always()
run: | run: |
docker image prune -a --force 2>&1 | tail -3 || true docker image prune -a --force 2>&1 | tail -3 || true
cargo clean || true cargo clean || true
+263
View File
@@ -0,0 +1,263 @@
# 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
+217
View File
@@ -0,0 +1,217 @@
# 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)
+191
View File
@@ -0,0 +1,191 @@
# 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.
+6 -297
View File
@@ -1,17 +1,11 @@
//! Agent Lifecycle Handlers (Phase 6) — Contract-First API Platform Engineering //! Agent Lifecycle Handlers (Phase 6)
//!
//! 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 actix_web::{web, HttpRequest, HttpResponse};
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
use std::sync::Arc; use std::sync::Arc;
use uuid::Uuid;
use chrono::Utc;
use crate::agent::{Agent, AgentConfig, AgentCapability, DefaultAgent}; use crate::agent::{Agent, AgentConfig, AgentCapability, DefaultAgent};
use crate::agent::client_sdk::SynthesisClient; use crate::agent::client_sdk::SynthesisClient;
use crate::handlers::response_builder; use crate::handlers::response_builder;
use mem_store::agent_repo::{AgentRepository, AgentPrompt, AgentSkill, AgentDecision, RolePromptMapping};
use tracing::{debug, info, error, warn}; use tracing::{debug, info, error, warn};
/// Register agent request /// Register agent request
@@ -86,50 +80,7 @@ pub async fn register_agent_handler(
metadata: std::collections::HashMap::new(), metadata: std::collections::HashMap::new(),
}; };
// Persist agent config to database via agent_registry table // Store agent config (stub: would persist to DB)
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))
.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); let agent = DefaultAgent::new(config);
// Extract JWT from request for agent reasoning calls // Extract JWT from request for agent reasoning calls
@@ -139,7 +90,7 @@ pub async fn register_agent_handler(
warn!("Agent registered without JWT token"); warn!("Agent registered without JWT token");
} }
info!("Agent registered and persisted: {}", agent.config().agent_id); info!("Agent registered: {}", agent.config().agent_id);
// Wire Temporal workflow (via api.riotpiao.com/workflow) // Wire Temporal workflow (via api.riotpiao.com/workflow)
// Temporal activities will: // Temporal activities will:
@@ -181,6 +132,8 @@ pub async fn register_agent_handler(
let workflow_id = data.get("workflow_id").and_then(|v| v.as_str()).unwrap_or("unknown"); 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"); 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); info!("Agent workflow started: workflow_id={}, run_id={}", workflow_id, run_id);
debug!("Temporal activity will persist agent state + reasoning traces"); debug!("Temporal activity will persist agent state + reasoning traces");
} }
@@ -198,7 +151,7 @@ pub async fn register_agent_handler(
capabilities: body.capabilities.clone(), capabilities: body.capabilities.clone(),
webhook_url: body.webhook_url.clone(), webhook_url: body.webhook_url.clone(),
rate_limit: agent.config().rate_limit, rate_limit: agent.config().rate_limit,
created_at: Utc::now().to_rfc3339(), created_at: chrono::Utc::now().to_rfc3339(),
status: "active".to_string(), status: "active".to_string(),
}) })
} }
@@ -364,247 +317,3 @@ 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")
}
}
}
+2 -18
View File
@@ -440,9 +440,6 @@ 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::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}", 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("/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); tracing::info!("HttpServer instance created, binding to 0.0.0.0:{}", port);
@@ -530,19 +527,8 @@ pub async fn ingest_handler(
INGEST_BYTES_TOTAL.inc_by(byte_count as u64); INGEST_BYTES_TOTAL.inc_by(byte_count as u64);
INGEST_RECORDS_TOTAL.inc_by(body.records.len() 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 // Execute ingest
let resp = execute_ingest(&state, &body, x_forward_user).await; let resp = execute_ingest(&state, &body).await;
INGEST_IN_FLIGHT.dec(); INGEST_IN_FLIGHT.dec();
resp resp
} }
@@ -551,7 +537,6 @@ pub async fn ingest_handler(
async fn execute_ingest( async fn execute_ingest(
state: &web::Data<AppState>, state: &web::Data<AppState>,
body: &IngestRequest, body: &IngestRequest,
x_forward_user: Option<String>,
) -> HttpResponse { ) -> HttpResponse {
let records: Vec<(String, String)> = body.records let records: Vec<(String, String)> = body.records
.iter() .iter()
@@ -582,9 +567,8 @@ async fn execute_ingest(
let worker = state.ingest_worker.clone(); let worker = state.ingest_worker.clone();
let project = body.project.clone(); let project = body.project.clone();
let ingest_id = body.ingest_id.clone(); let ingest_id = body.ingest_id.clone();
let x_fwd = x_forward_user.clone();
tokio::spawn(async move { tokio::spawn(async move {
if let Err(e) = worker.process_ingest_with_auth(&project, &ingest_id, records, x_fwd).await { if let Err(e) = worker.process_ingest(&project, &ingest_id, records).await {
tracing::error!("Ingest failed: {}", e); tracing::error!("Ingest failed: {}", e);
} }
}); });
+21 -139
View File
@@ -68,201 +68,83 @@ impl IngestWorker {
ingest_id: &str, ingest_id: &str,
records: Vec<(String, String)>, // (content, source) records: Vec<(String, String)>, // (content, source)
) -> Result<()> { ) -> Result<()> {
self.process_ingest_with_auth(project, ingest_id, records, None).await tracing::info!("Processing ingest: project={}, id={}, records={}", project, ingest_id, records.len());
}
/// 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 // Update job status to processing
if let Err(e) = sqlx::query("UPDATE ingest_jobs SET status=$1, started_at=NOW() WHERE ingest_id=$2") sqlx::query("UPDATE ingest_jobs SET status=$1, started_at=NOW() WHERE ingest_id=$2")
.bind("processing") .bind("processing")
.bind(ingest_id) .bind(ingest_id)
.execute(&self.pool) .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_entities = 0;
let mut total_edges = 0; let mut total_edges = 0;
let mut total_reviews = 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 // Process each record through the ingest pipeline
for (idx, (content, source)) in records.iter().enumerate() { 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 // Create episode from record
let episode = Episode { let episode = Episode {
id: record_id.clone(), id: format!("{}-{}", ingest_id, idx),
project_id: project.to_string(), project_id: project.to_string(),
text: content.clone(), text: content.clone(),
wiki_links: extract_wiki_links(content), wiki_links: extract_wiki_links(content),
}; };
// Run extraction pipeline (entity + fact extraction + contradiction detection) // Run extraction pipeline (entity + fact extraction + contradiction detection)
let x_forward_user_ref = x_forward_user.as_deref(); match self.pipeline.ingest(&episode).await {
match self.pipeline.ingest_with_auth(&episode, x_forward_user_ref).await {
Ok(result) => { Ok(result) => {
tracing::debug!( tracing::debug!(
target: "ingest", "Pipeline extracted {} entities, {} edges for episode {}",
record_id = %record_id, result.entities.len(),
entity_count = result.entities.len(), result.edges.len(),
edge_count = result.edges.len(), episode.id
review_count = result.reviews.len(),
"Pipeline extraction successful"
); );
// Save entities to database (normally via EntityRepo, using direct SQL for now) // Save entities to database (normally via EntityRepo, using direct SQL for now)
for entity in &result.entities { for entity in &result.entities {
match save_entity_to_db(&self.pool, entity).await { if let Err(e) = save_entity_to_db(&self.pool, entity).await {
Ok(_) => { tracing::warn!("Failed to save entity {}: {}", entity.name, e);
tracing::debug!( } else {
target: "ingest",
record_id = %record_id,
entity_name = &entity.name,
entity_type = entity.entity_type.as_str(),
"Saved entity"
);
total_entities += 1; 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) // Save edges to database (normally via EdgeRepo, using direct SQL for now)
for edge in &result.edges { for edge in &result.edges {
match save_edge_to_db(&self.pool, edge).await { if let Err(e) = save_edge_to_db(&self.pool, edge).await {
Ok(_) => { tracing::warn!("Failed to save edge: {}", e);
tracing::debug!( } else {
target: "ingest",
record_id = %record_id,
relation_type = &edge.relation_type,
"Saved edge"
);
total_edges += 1; 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(); total_reviews += result.reviews.len();
} }
Err(e) => { Err(e) => {
let msg = format!("Record {}: {}", record_id, e); tracing::error!("Pipeline failed for episode {}: {}", episode.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 // Continue processing other records
} }
} }
} }
// Mark job complete // Mark job complete
let final_status = if extraction_errors.is_empty() && save_errors.is_empty() { sqlx::query("UPDATE ingest_jobs SET status=$1, completed_at=NOW() WHERE ingest_id=$2")
"done" .bind("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) .bind(ingest_id)
.execute(&self.pool) .execute(&self.pool)
.await .await?;
{
tracing::error!(
target: "ingest",
error = %e,
ingest_id = ingest_id,
"Failed to update job completion status"
);
}
tracing::info!( tracing::info!(
target: "ingest", target: "observability",
event = "ingest_complete", event = "ingest_complete",
ingest_id = ingest_id, ingest_id = ingest_id,
project = project,
entities = total_entities, entities = total_entities,
edges = total_edges, edges = total_edges,
reviews = total_reviews, reviews = total_reviews,
extraction_errors = extraction_errors.len(), "Ingest completed"
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(()) Ok(())
} }
-1
View File
@@ -2,7 +2,6 @@
/// ///
/// These structures attach to Entity via entity_type discriminator. /// These structures attach to Entity via entity_type discriminator.
/// AgentPrompt, AgentSkill, AgentDecision each carry domain-specific /// 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. /// fields that enable the agent to learn from its own behavior.
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
-1
View File
@@ -1,6 +1,5 @@
/// Community domain model for temporal graph-RAG. /// Community domain model for temporal graph-RAG.
/// Single Responsibility: Community (cluster) storage and metadata. /// Single Responsibility: Community (cluster) storage and metadata.
#[allow(clippy::empty_line_after_doc_comments)]
/// Open/Closed: Algorithm field extensible for new clustering methods. /// Open/Closed: Algorithm field extensible for new clustering methods.
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
-2
View File
@@ -1,6 +1,5 @@
/// Edge domain model for temporal graph-RAG. /// Edge domain model for temporal graph-RAG.
/// Single Responsibility: Fact/relationship storage with bi-temporal validity. /// Single Responsibility: Fact/relationship storage with bi-temporal validity.
#[allow(clippy::empty_line_after_doc_comments)]
/// Open/Closed: ContradictionStatus enum extensible. /// Open/Closed: ContradictionStatus enum extensible.
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
@@ -30,7 +29,6 @@ impl ContradictionStatus {
} }
} }
#[allow(clippy::should_implement_trait)]
pub fn from_str(s: &str) -> Self { pub fn from_str(s: &str) -> Self {
match s.to_lowercase().as_str() { match s.to_lowercase().as_str() {
"active" => Self::Active, "active" => Self::Active,
-2
View File
@@ -1,7 +1,6 @@
/// Entity domain model for temporal graph-RAG. /// Entity domain model for temporal graph-RAG.
/// Single Responsibility: Entity identity and metadata. /// Single Responsibility: Entity identity and metadata.
/// Open/Closed: EntityType enum extensible. /// Open/Closed: EntityType enum extensible.
#[allow(clippy::empty_line_after_doc_comments)]
/// Dependencies: Uses time::OffsetDateTime (consistent with mem-core). /// Dependencies: Uses time::OffsetDateTime (consistent with mem-core).
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
@@ -44,7 +43,6 @@ impl EntityType {
} }
} }
#[allow(clippy::should_implement_trait)]
pub fn from_str(s: &str) -> Self { pub fn from_str(s: &str) -> Self {
match s.to_lowercase().as_str() { match s.to_lowercase().as_str() {
"person" => Self::Person, "person" => Self::Person,
+2 -1
View File
@@ -135,10 +135,11 @@ pub fn run_loop(
#[cfg(test)] #[cfg(test)]
mod tests { mod tests {
use super::*;
#[test] #[test]
fn test_loop_basic() { fn test_loop_basic() {
// Placeholder test to verify it compiles // Placeholder test to verify it compiles
assert!(true);
} }
} }
+7 -7
View File
@@ -403,7 +403,7 @@ pub fn lookup(sig: &Signature, lessons: &[Lesson], floor: f32) -> Option<Hit> {
let mut best: Option<(f32, &Lesson)> = None; let mut best: Option<(f32, &Lesson)> = None;
for l in lessons.iter().filter(|l| l.tool == sig.tool) { for l in lessons.iter().filter(|l| l.tool == sig.tool) {
let s = similarity(&sig.normalised, &l.normalised); let s = similarity(&sig.normalised, &l.normalised);
if s >= floor && best.is_none_or(|(bs, _)| s > bs) { if s >= floor && best.map_or(true, |(bs, _)| s > bs) {
best = Some((s, l)); best = Some((s, l));
} }
} }
@@ -503,7 +503,7 @@ pub fn tool_of_cmd(cmd: &str) -> String {
"kubectl" | "k" => "kubectl".into(), "kubectl" | "k" => "kubectl".into(),
"docker" | "podman" => "docker".into(), "docker" | "podman" => "docker".into(),
"terraform" | "tofu" => "terraform".into(), "terraform" | "tofu" => "terraform".into(),
"" => "unknown".into(), other if other.is_empty() => "unknown".into(),
other => other.to_string(), 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"); s.push_str("`confirmed`, which outranks inferred lessons at equal similarity.\n\n");
let mut sorted: Vec<&Lesson> = lessons.iter().collect(); let mut sorted: Vec<&Lesson> = lessons.iter().collect();
sorted.sort_by_key(|a| std::cmp::Reverse(a.seen)); sorted.sort_by(|a, b| b.seen.cmp(&a.seen));
for l in sorted { for l in sorted {
s.push_str(&format!("## {}\n\n", l.raw.trim())); 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", "- seen: {} | last: {} | confidence: {:?}\n",
l.seen, l.last_seen, l.confidence l.seen, l.last_seen, l.confidence
)); ));
s.push_str(&format!("- signature: `{}`\n", &l.sig_sha[..12])); s.push_str(&format!("- signature: `{}`\n", l.sig_sha[..12].to_string()));
s.push_str("- resolved by:\n"); s.push_str("- resolved by:\n");
for r in &l.resolution { for r in &l.resolution {
s.push_str(&format!(" ```\n {r}\n ```\n")); s.push_str(&format!(" ```\n {r}\n ```\n"));
@@ -712,7 +712,7 @@ mod tests {
ev("t2", "npm pkg set overrides.react=19", 0, ""), ev("t2", "npm pkg set overrides.react=19", 0, ""),
ev("t3", "npm ci", 0, "ok"), ev("t3", "npm ci", 0, "ok"),
]; ];
let ls = derive_lessons(&events, tool_of_cmd); let ls = derive_lessons(&events, |c| tool_of_cmd(c));
assert_eq!(ls.len(), 1); assert_eq!(ls.len(), 1);
assert_eq!(ls[0].resolution, vec!["npm pkg set overrides.react=19"]); assert_eq!(ls[0].resolution, vec!["npm pkg set overrides.react=19"]);
assert_eq!(ls[0].confidence, Confidence::Inferred); assert_eq!(ls[0].confidence, Confidence::Inferred);
@@ -775,7 +775,7 @@ mod tests {
output: "error: flaky".into(), output: "error: flaky".into(),
}; };
let events = vec![ev("npm ci", 1), ev("npm ci", 0)]; let events = vec![ev("npm ci", 1), ev("npm ci", 0)];
assert!(derive_lessons(&events, tool_of_cmd).is_empty()); assert!(derive_lessons(&events, |c| tool_of_cmd(c)).is_empty());
} }
#[test] #[test]
@@ -798,7 +798,7 @@ mod tests {
sig_sha: "abc".into(), sig_sha: "abc".into(),
rule: "r".into(), rule: "r".into(),
}; };
assert_eq!(lookup(&exact, std::slice::from_ref(&l), 0.5).unwrap().tier, Tier::Exact); assert_eq!(lookup(&exact, &[l.clone()], 0.5).unwrap().tier, Tier::Exact);
let unrelated = Signature { let unrelated = Signature {
tool: "npm".into(), tool: "npm".into(),
+2 -2
View File
@@ -152,11 +152,11 @@ impl FormatHandler for CsvFormatter {
async fn format(&self, result: &OptimizationResult) -> Result<Vec<u8>, String> { async fn format(&self, result: &OptimizationResult) -> Result<Vec<u8>, String> {
let output = format!( let output = format!(
"{},{},{},{:.2}\n", "{},{},{},{}\n",
escape_csv(&result.plugin), escape_csv(&result.plugin),
result.original.len(), result.original.len(),
result.optimized.len(), result.optimized.len(),
result.ratio format!("{:.2}", result.ratio)
); );
Ok(output.into_bytes()) Ok(output.into_bytes())
} }
+2 -2
View File
@@ -40,7 +40,7 @@ impl CcrStore {
// Remove oldest entry if at capacity // Remove oldest entry if at capacity
if cache.len() >= self.max_entries { if cache.len() >= self.max_entries {
if let Some(oldest_key) = cache.keys().next().cloned() { if let Some(oldest_key) = cache.keys().next().cloned() {
cache.swap_remove(&oldest_key); cache.remove(&oldest_key);
} }
} }
@@ -57,7 +57,7 @@ impl CcrStore {
// Check if expired // Check if expired
let duration = OffsetDateTime::now_utc() - *timestamp; let duration = OffsetDateTime::now_utc() - *timestamp;
if duration.whole_seconds() > self.ttl_secs as i64 { if duration.whole_seconds() > self.ttl_secs as i64 {
cache.swap_remove(hash); cache.remove(hash);
return Ok(None); return Ok(None);
} }
+5 -5
View File
@@ -7,7 +7,7 @@
//! - Drop: redundant homogeneous elements, long string values //! - Drop: redundant homogeneous elements, long string values
use anyhow::Result; use anyhow::Result;
use serde_json::Value; use serde_json::{json, Value};
use std::collections::HashMap; use std::collections::HashMap;
pub struct JsonCrusher; pub struct JsonCrusher;
@@ -45,8 +45,8 @@ impl JsonCrusher {
let mut result = Vec::new(); let mut result = Vec::new();
// Add start items // Add start items
for item in items.iter().take(start_count.min(len)) { for i in 0..start_count.min(len) {
result.push(item.clone()); result.push(items[i].clone());
} }
// Select mid-array items by variance/importance // Select mid-array items by variance/importance
@@ -58,8 +58,8 @@ impl JsonCrusher {
// Add end items // Add end items
if end_count > 0 { if end_count > 0 {
for item in items.iter().skip(len.saturating_sub(end_count)) { for i in (len - end_count)..len {
result.push(item.clone()); result.push(items[i].clone());
} }
} }
@@ -5,7 +5,7 @@
use super::plugin::OptimizerService; use super::plugin::OptimizerService;
use crate::prompt::CacheMetrics; use crate::prompt::CacheMetrics;
use crate::domain::Chunk; use crate::domain::{Chunk, Record};
use anyhow::Result; use anyhow::Result;
/// Query optimizer: compresses chunks before LLM processing /// Query optimizer: compresses chunks before LLM processing
@@ -83,7 +83,7 @@ impl QueryOptimizer {
match service.optimize(&chunk_text, &content_type, Some("raw")).await { match service.optimize(&chunk_text, &content_type, Some("raw")).await {
Ok(bytes) => { Ok(bytes) => {
let text = String::from_utf8(bytes) let text = String::from_utf8(bytes)
.unwrap_or(chunk_text); .unwrap_or_else(|_| chunk_text);
Ok(text) Ok(text)
} }
Err(_) => { Err(_) => {
+1 -1
View File
@@ -42,7 +42,7 @@ impl ContentRouter {
/// Check if content is valid JSON /// Check if content is valid JSON
fn is_json(content: &str) -> bool { fn is_json(content: &str) -> bool {
let trimmed = content.trim(); let trimmed = content.trim();
if !(trimmed.starts_with('{') || trimmed.starts_with('[')) { if !((trimmed.starts_with('{') || trimmed.starts_with('['))) {
return false; return false;
} }
serde_json::from_str::<serde_json::Value>(trimmed).is_ok() serde_json::from_str::<serde_json::Value>(trimmed).is_ok()
+1 -1
View File
@@ -128,7 +128,7 @@ impl TextCompressor {
} }
// Capitalization (usually proper nouns or emphatic) // Capitalization (usually proper nouns or emphatic)
if token.chars().next().is_some_and(|c| c.is_uppercase()) && token.len() > 1 { if token.chars().next().map_or(false, |c| c.is_uppercase()) && token.len() > 1 {
score += 1.0; score += 1.0;
} }
+2 -4
View File
@@ -12,9 +12,7 @@ const CACHE_TURN: &str = include_str!("../../../templates/gru-mem-turn.txt");
const BUDGET_TOTAL: usize = 32768; const BUDGET_TOTAL: usize = 32768;
const BUDGET_RESPONSE: usize = 2048; const BUDGET_RESPONSE: usize = 2048;
#[allow(dead_code)]
const BUDGET_SYSTEM: usize = 400; const BUDGET_SYSTEM: usize = 400;
#[allow(dead_code)]
const BUDGET_QUESTION: usize = 150; const BUDGET_QUESTION: usize = 150;
const BUDGET_MEMORY_MAX: usize = 1024; const BUDGET_MEMORY_MAX: usize = 1024;
const BUDGET_CHUNK_MAX: usize = 5000; const BUDGET_CHUNK_MAX: usize = 5000;
@@ -370,7 +368,7 @@ fn estimate_tokens(text: &str) -> usize {
#[cfg(test)] #[cfg(test)]
mod tests { mod tests {
use super::*; use super::*;
use crate::domain::{Chunk, Record, Role, Provenance}; use crate::domain::{Chunk, Record, Role, Provenance, Level};
use time::OffsetDateTime; use time::OffsetDateTime;
fn make_test_chunk(text: &str) -> Chunk { fn make_test_chunk(text: &str) -> Chunk {
@@ -647,7 +645,7 @@ mod tests {
let metrics = result.unwrap(); let metrics = result.unwrap();
let ratio = metrics.compression_ratio(); let ratio = metrics.compression_ratio();
assert!((0.0..=100.0).contains(&ratio)); assert!(ratio >= 0.0 && ratio <= 100.0);
} }
#[test] #[test]
+2
View File
@@ -1,5 +1,7 @@
use crate::domain::{ProjectId, QueryId};
use anyhow::{anyhow, Result}; use anyhow::{anyhow, Result};
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::path::Path; use std::path::Path;
/// A single standing query. /// A single standing query.
+1 -7
View File
@@ -1,4 +1,4 @@
use crate::Level; use crate::{Level, Query};
use anyhow::Result; use anyhow::Result;
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
@@ -17,12 +17,6 @@ pub struct QueryExecutor {
// For now: proof-of-concept with mock data // For now: proof-of-concept with mock data
} }
impl Default for QueryExecutor {
fn default() -> Self {
Self::new()
}
}
impl QueryExecutor { impl QueryExecutor {
/// Create executor. /// Create executor.
pub fn new() -> Self { pub fn new() -> Self {
+3 -2
View File
@@ -71,10 +71,11 @@ impl QueryLevels {
} }
// Check level filter // Check level filter
if !self.level_filter.is_empty() if !self.level_filter.is_empty() {
&& !self.level_filter.contains(&level.to_string()) { if !self.level_filter.contains(&level.to_string()) {
return false; return false;
} }
}
// Check evidence/reference flags // Check evidence/reference flags
if level == "R" { if level == "R" {
-15
View File
@@ -6,7 +6,6 @@
/// - Single Responsibility: each scorer does one thing /// - Single Responsibility: each scorer does one thing
/// - Open/Closed: add new scorers without modifying existing /// - Open/Closed: add new scorers without modifying existing
/// - Liskov Substitution: all scorers implement DocumentScorer /// - Liskov Substitution: all scorers implement DocumentScorer
#[allow(clippy::empty_line_after_doc_comments)]
/// - Dependency Inversion: depend on trait, not concrete types /// - Dependency Inversion: depend on trait, not concrete types
use anyhow::Result; use anyhow::Result;
@@ -54,7 +53,6 @@ impl DocumentScorer for GlobalTfIdfScorer {
} }
/// Project-scoped TF-IDF Scorer: scoring within project boundaries /// Project-scoped TF-IDF Scorer: scoring within project boundaries
#[allow(dead_code)]
pub struct ProjectTfIdfScorer { pub struct ProjectTfIdfScorer {
project: String, project: String,
vocabulary: Arc<std::collections::BTreeMap<String, f32>>, vocabulary: Arc<std::collections::BTreeMap<String, f32>>,
@@ -95,18 +93,11 @@ impl DocumentScorer for ProjectTfIdfScorer {
} }
/// Semantic Scorer: vector similarity (placeholder) /// Semantic Scorer: vector similarity (placeholder)
#[allow(dead_code)]
pub struct SemanticScorer { pub struct SemanticScorer {
_embeddings_client: Arc<()>, // Placeholder _embeddings_client: Arc<()>, // Placeholder
_pgvector: Arc<()>, // Placeholder _pgvector: Arc<()>, // Placeholder
} }
impl Default for SemanticScorer {
fn default() -> Self {
Self::new()
}
}
impl SemanticScorer { impl SemanticScorer {
pub fn new() -> Self { pub fn new() -> Self {
Self { Self {
@@ -165,12 +156,6 @@ pub struct ScoringPipeline {
scorers: Vec<(String, f32, Arc<dyn DocumentScorer>)>, // name, weight, scorer scorers: Vec<(String, f32, Arc<dyn DocumentScorer>)>, // name, weight, scorer
} }
impl Default for ScoringPipeline {
fn default() -> Self {
Self::new()
}
}
impl ScoringPipeline { impl ScoringPipeline {
pub fn new() -> Self { pub fn new() -> Self {
Self { Self {
+1 -2
View File
@@ -81,7 +81,6 @@ impl SymptomVector {
/// Internal structure for tokens during extraction /// Internal structure for tokens during extraction
#[derive(Debug, Clone)] #[derive(Debug, Clone)]
#[allow(dead_code)]
struct SymptomTokens { struct SymptomTokens {
keywords: Vec<String>, keywords: Vec<String>,
error_codes: Vec<String>, error_codes: Vec<String>,
@@ -393,7 +392,7 @@ mod tests {
let words: Vec<&str> = symptom.normalised.split_whitespace().collect(); let words: Vec<&str> = symptom.normalised.split_whitespace().collect();
for word in &words { for word in &words {
// Check if this word is a stop word // 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 // Should contain key terms
assert!(symptom.normalised.contains("resolve")); assert!(symptom.normalised.contains("resolve"));
+4 -2
View File
@@ -267,9 +267,11 @@ fn test_compression_handles_large_content() {
fn test_multi_chunk_search_consistency() { fn test_multi_chunk_search_consistency() {
let optimizer = ContextOptimizer::new().expect("optimizer init"); let optimizer = ContextOptimizer::new().expect("optimizer init");
let chunks = ["ERROR: connection failed\nDEBUG: thread id=100", let chunks = vec![
"ERROR: connection failed\nDEBUG: thread id=100",
"ERROR: timeout after 5000ms\nTRACE: stack unwinding", "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 let optimized_chunks: Vec<_> = chunks
.iter() .iter()
+3 -1
View File
@@ -196,6 +196,7 @@ fn gate_memory_bounded() {
// Should not panic from memory exhaustion // Should not panic from memory exhaustion
// If we get here, we passed the gate // If we get here, we passed the gate
assert!(true, "memory usage bounded");
} }
#[test] #[test]
@@ -230,7 +231,7 @@ fn gate_compression_targets_met() {
]; ];
for (content, name, min_compression) in fixtures.iter() { for (content, name, min_compression) in fixtures.iter() {
let optimized = optimizer.optimize(content).unwrap_or_else(|_| panic!("optimize {}", name)); let optimized = optimizer.optimize(content).expect(&format!("optimize {}", name));
let ratio = optimized.compressed.len() as f32 / content.len() as f32; let ratio = optimized.compressed.len() as f32 / content.len() as f32;
// At least some compression should happen // At least some compression should happen
@@ -331,4 +332,5 @@ fn gate_summary_report() {
println!("\n🚀 STATUS: M3.8 READY FOR PRODUCTION"); println!("\n🚀 STATUS: M3.8 READY FOR PRODUCTION");
assert!(true); // Just for testing framework
} }
@@ -83,7 +83,6 @@ impl ContradictionPreFilter {
/// LLM-based contradiction detector (stage 2) /// LLM-based contradiction detector (stage 2)
/// Only called if pre-filter returns true (cost optimization) /// Only called if pre-filter returns true (cost optimization)
#[allow(dead_code)]
pub struct LlmContradictionDetector { pub struct LlmContradictionDetector {
model_name: String, model_name: String,
auto_confirm_threshold: f32, auto_confirm_threshold: f32,
+21 -131
View File
@@ -44,15 +44,10 @@ impl ExtractedEntity {
#[async_trait] #[async_trait]
pub trait EntityExtractor: Send + Sync { pub trait EntityExtractor: Send + Sync {
async fn extract(&self, text: &str) -> Result<Vec<ExtractedEntity>>; 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) /// LLM-based extractor with reflection verification (stage 1 + 2)
/// Uses Authentik JWT tokens for authentication to LLM gateway /// Uses Authentik JWT tokens for authentication to LLM gateway
#[allow(dead_code)]
pub struct LlmEntityExtractor { pub struct LlmEntityExtractor {
model_name: String, model_name: String,
enable_reflection: bool, enable_reflection: bool,
@@ -125,42 +120,29 @@ impl LlmEntityExtractor {
Ok(parsed.verified.into_iter().map(|v| (v.name, v.present)).collect()) Ok(parsed.verified.into_iter().map(|v| (v.name, v.present)).collect())
} }
/// Call LLM via api.riotpiao.com using X-Forward-User auth/exchange /// Call LLM via api.riotpiao.com using Authentik JWT
/// Supports: Authentik JWT, X-Forward-User header, or API key fallback /// Token is fetched from Authentik service account and cached
async fn call_llm_endpoint(&self, prompt: &str, x_forward_user: Option<&str>) -> Result<String> { async fn call_llm_endpoint(&self, prompt: &str) -> Result<String> {
let endpoint = std::env::var("LLM_ENDPOINT") let endpoint = std::env::var("LLM_ENDPOINT")
.unwrap_or_else(|_| "http://api-internal.riotpiao.com:8000/v1/chat/completions".to_string()); .unwrap_or_else(|_| "http://api-internal.riotpiao.com:8000/v1/chat/completions".to_string());
let model = std::env::var("LLM_MODEL") let model = std::env::var("LLM_MODEL")
.unwrap_or_else(|_| "qwen:7b".to_string()); .unwrap_or_else(|_| "qwen:7b".to_string());
// Get auth header: prefer X-Forward-User, fallback to Authentik JWT, then API key // Get JWT token from Authentik
let auth_header = if let Some(user) = x_forward_user { let auth_header = if let Some(jwt_issuer) = &self.jwt_issuer {
// 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; let issuer = jwt_issuer.lock().await;
match issuer.get_access_token().await { match issuer.get_access_token().await {
Ok(token) => { Ok(token) => format!("Bearer {}", token),
tracing::info!("Using Authentik JWT for LLM auth");
format!("Bearer {}", token)
},
Err(e) => { Err(e) => {
tracing::warn!("Failed to get Authentik JWT: {}", e); tracing::warn!("Failed to get Authentik JWT: {}", e);
// Fallback to env var return Err(e);
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 { } else {
// Fallback to env var if Authentik not configured // Fallback to env var if Authentik not configured
let api_key = std::env::var("LLM_API_KEY") let api_key = std::env::var("LLM_API_KEY")
.or_else(|_| std::env::var("MEM_API_KEY")) .or_else(|_| std::env::var("MEM_API_KEY"))
.unwrap_or_else(|_| "test-key".to_string()); .unwrap_or_else(|_| "default-key".to_string());
tracing::info!("Using LLM_API_KEY for LLM auth");
format!("Bearer {}", api_key) format!("Bearer {}", api_key)
}; };
@@ -177,33 +159,23 @@ impl LlmEntityExtractor {
"max_tokens": 12000 "max_tokens": 12000
}); });
let mut request = client let response = client
.post(&endpoint) .post(&endpoint)
.header("Content-Type", "application/json"); .header("Authorization", auth_header)
.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) .json(&payload)
.timeout(std::time::Duration::from_secs(90)) .timeout(std::time::Duration::from_secs(90))
.send() .send()
.await?; .await?;
let status = response.status(); if !response.status().is_success() {
if !status.is_success() { tracing::warn!(
let error_text = response.text().await.unwrap_or_default();
tracing::error!(
"LLM API error: {} - {}", "LLM API error: {} - {}",
status, response.status(),
error_text response.text().await.unwrap_or_default()
); );
// Return error instead of silently returning empty array // Fallback to mock response on error
return Err(anyhow::anyhow!("LLM API failed with status {}: {}", status, error_text)); return Ok(r#"{"entities": []}"#.to_string());
} }
let data: serde_json::Value = response.json().await?; let data: serde_json::Value = response.json().await?;
@@ -285,10 +257,7 @@ Respond in JSON:
// Try real LLM first, fallback to mock if not configured // Try real LLM first, fallback to mock if not configured
let extraction_response = if std::env::var("LLM_ENDPOINT").is_ok() { let extraction_response = if std::env::var("LLM_ENDPOINT").is_ok() {
self.call_llm_endpoint(&prompt, None).await.unwrap_or_else(|e| { self.call_llm_endpoint(&prompt).await.unwrap_or_else(|_| self.simulate_llm(&prompt).unwrap_or_default())
tracing::error!("LLM entity extraction failed: {}, using mock", e);
self.simulate_llm(&prompt).unwrap_or_default()
})
} else { } else {
self.simulate_llm(&prompt)? self.simulate_llm(&prompt)?
}; };
@@ -313,7 +282,7 @@ Respond in JSON:
); );
let reflection = if std::env::var("LLM_ENDPOINT").is_ok() { let reflection = if std::env::var("LLM_ENDPOINT").is_ok() {
self.call_llm_endpoint(&reflection_prompt, None).await.unwrap_or_else(|e| { self.call_llm_endpoint(&reflection_prompt).await.unwrap_or_else(|e| {
tracing::warn!("Reflection LLM call failed: {}, skipping verification", e); tracing::warn!("Reflection LLM call failed: {}, skipping verification", e);
String::new() String::new()
}) })
@@ -343,85 +312,6 @@ Respond in JSON:
Ok(entities) 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) /// Fallback extractor: Use wiki_links if LLM fails (stage 3)
@@ -440,7 +330,7 @@ impl EntityExtractor for WikiLinkFallbackExtractor {
entities.push(ExtractedEntity { entities.push(ExtractedEntity {
name: name_str.to_string(), name: name_str.to_string(),
entity_type: EntityType::Unknown, entity_type: EntityType::Unknown,
summary: "Mentioned in episode".to_string(), summary: format!("Mentioned in episode"),
confidence: 0.7, // Lower confidence for fallback confidence: 0.7, // Lower confidence for fallback
}); });
} }
@@ -528,6 +418,6 @@ mod tests {
let text = "[[Entity1]] and [[Entity2]]"; let text = "[[Entity1]] and [[Entity2]]";
let entities = composite.extract(text).await.unwrap(); let entities = composite.extract(text).await.unwrap();
assert!(!entities.is_empty()); assert!(entities.len() > 0);
} }
} }
+4 -1
View File
@@ -10,7 +10,10 @@
use anyhow::Result; use anyhow::Result;
use async_trait::async_trait; use async_trait::async_trait;
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
use tracing::debug; use std::collections::HashMap;
use tracing::{debug, info};
use mem_core::entity::Entity;
use mem_core::edge::Edge;
/// Memorability decision for entity or fact /// Memorability decision for entity or fact
#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq)] #[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq)]
+2 -8
View File
@@ -59,15 +59,10 @@ impl IngestPipeline {
/// Execute extraction pipeline for episode /// Execute extraction pipeline for episode
/// CRAP: 14 (Low: orchestration only, delegates to stages) /// CRAP: 14 (Low: orchestration only, delegates to stages)
pub async fn ingest(&self, episode: &Episode) -> Result<ExtractionResult> { 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); debug!("Starting ingest for episode: {}", episode.id);
// Stage 1: Extract entities (with optional auth header) // Stage 1: Extract entities
let extracted_entities = self.entity_extractor.extract_with_auth(&episode.text, x_forward_user).await?; let extracted_entities = self.entity_extractor.extract(&episode.text).await?;
debug!("Extracted {} entities", extracted_entities.len()); debug!("Extracted {} entities", extracted_entities.len());
// Convert to domain entities // Convert to domain entities
@@ -149,7 +144,6 @@ impl IngestPipeline {
/// Async queue worker: Process episodes from queue /// Async queue worker: Process episodes from queue
/// CRAP: 12 (Async loop, straightforward) /// CRAP: 12 (Async loop, straightforward)
#[allow(dead_code)]
pub struct QueueWorker { pub struct QueueWorker {
pipeline: Arc<IngestPipeline>, pipeline: Arc<IngestPipeline>,
batch_size: usize, batch_size: usize,
+2 -2
View File
@@ -14,7 +14,7 @@ use tracing::{debug, info};
use crate::grm_retriever::{ use crate::grm_retriever::{
EntityContext, FactContext, GraphContextRetriever, MemorabilityDecision, GrmConfig, MockGrmRetriever, EntityContext, FactContext, GraphContextRetriever, MemorabilityDecision, GrmConfig, MockGrmRetriever,
}; };
use mem_core::entity::Entity; use mem_core::entity::{Entity, EntityType};
use mem_core::edge::Edge; use mem_core::edge::Edge;
/// Entity filtering result /// Entity filtering result
@@ -88,7 +88,7 @@ impl MemorabilityGate {
let (filtered, reason) = match context.decision { let (filtered, reason) = match context.decision {
MemorabilityDecision::Keep => { MemorabilityDecision::Keep => {
if context.matched_entity_id.is_some() { if context.matched_entity_id.is_some() {
(true, "Existing entity (merge required)".to_string()) (true, format!("Existing entity (merge required)"))
} else { } else {
(false, format!("New entity (score: {:.2})", context.memorability_score)) (false, format!("New entity (score: {:.2})", context.memorability_score))
} }
+1 -5
View File
@@ -20,7 +20,6 @@ pub struct RefMetadata {
} }
/// Obsidian REST API client /// Obsidian REST API client
#[allow(dead_code)]
pub struct ObsidianClient { pub struct ObsidianClient {
base_url: String, base_url: String,
} }
@@ -48,7 +47,6 @@ impl ObsidianClient {
} }
/// ObsidianRefSource: Fetches & chunks reference documents from Obsidian vault /// ObsidianRefSource: Fetches & chunks reference documents from Obsidian vault
#[allow(dead_code)]
pub struct ObsidianRefSource { pub struct ObsidianRefSource {
client: ObsidianClient, client: ObsidianClient,
project: String, project: String,
@@ -70,13 +68,11 @@ impl ObsidianRefSource {
} }
/// Check if a file path is allowed (matches configured prefixes) /// Check if a file path is allowed (matches configured prefixes)
#[allow(dead_code)]
fn is_allowed_path(&self, path: &str) -> bool { fn is_allowed_path(&self, path: &str) -> bool {
self.allowed_paths.iter().any(|prefix| path.starts_with(prefix)) self.allowed_paths.iter().any(|prefix| path.starts_with(prefix))
} }
/// Chunk reference document via heading-boundary logic /// Chunk reference document via heading-boundary logic
#[allow(dead_code)]
fn chunk_document(&self, path: &str, content: &str) -> Vec<Record> { fn chunk_document(&self, path: &str, content: &str) -> Vec<Record> {
// M3.6.1 heading-boundary chunking // M3.6.1 heading-boundary chunking
// - Split by headings // - Split by headings
@@ -207,7 +203,7 @@ mod tests {
let chunks = source.chunk_document("docs/test.md", content); let chunks = source.chunk_document("docs/test.md", content);
// Should split by headings // Should split by headings
assert!(!chunks.is_empty()); assert!(chunks.len() > 0);
} }
#[test] #[test]
+2 -1
View File
@@ -60,7 +60,8 @@ impl MetricsCollector {
self.by_project self.by_project
.lock() .lock()
.unwrap() .unwrap()
.get(project).cloned() .get(project)
.map(|m| m.clone())
} }
/// Get all project metrics. /// Get all project metrics.
+1 -1
View File
@@ -306,7 +306,7 @@ impl QueryMetricsRepository {
let mut repo = self.metrics.lock().unwrap(); let mut repo = self.metrics.lock().unwrap();
repo.get_mut(query_id) repo.get_mut(query_id)
.ok_or_else(|| format!("Query {} not found", query_id)) .ok_or_else(|| format!("Query {} not found", query_id))
.map(f) .map(|metrics| f(metrics))
} }
/// Get progress for a query /// Get progress for a query
+3 -5
View File
@@ -4,10 +4,9 @@
/// ///
/// Used to scope queries to project namespaces and enable graph traversal. /// Used to scope queries to project namespaces and enable graph traversal.
/// For example: poimen/tools/kubectl.md [[debugging.md]] creates an edge /// 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). /// from tools/kubectl to debugging (within same project).
use anyhow::Result; use anyhow::{anyhow, Result};
use regex::Regex; use regex::Regex;
use std::collections::{HashMap, HashSet}; use std::collections::{HashMap, HashSet};
use std::path::{Path, PathBuf}; use std::path::{Path, PathBuf};
@@ -80,7 +79,6 @@ impl WikiLinkParser {
} }
/// Graph Index: Stores and queries wiki-link relationships /// Graph Index: Stores and queries wiki-link relationships
#[allow(dead_code)]
pub struct WikiLinkGraph { pub struct WikiLinkGraph {
/// Forward links: source -> [targets] /// Forward links: source -> [targets]
forward_links: HashMap<String, Vec<String>>, forward_links: HashMap<String, Vec<String>>,
@@ -102,11 +100,11 @@ impl WikiLinkGraph {
/// Add a wiki-link edge /// Add a wiki-link edge
pub fn add_link(&mut self, source: &str, target: &str) { pub fn add_link(&mut self, source: &str, target: &str) {
self.forward_links.entry(source.to_string()) self.forward_links.entry(source.to_string())
.or_default() .or_insert_with(Vec::new)
.push(target.to_string()); .push(target.to_string());
self.backward_links.entry(target.to_string()) self.backward_links.entry(target.to_string())
.or_default() .or_insert_with(Vec::new)
.push(source.to_string()); .push(source.to_string());
} }
+4 -4
View File
@@ -34,7 +34,7 @@ pub enum AuthMode {
impl AuthMode { impl AuthMode {
/// Detect from base URL or explicit env var. /// 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() { if api_key.is_empty() {
return Self::None; return Self::None;
} }
@@ -87,7 +87,6 @@ struct Choice {
} }
#[derive(Debug, Deserialize)] #[derive(Debug, Deserialize)]
#[allow(dead_code)]
struct MessageResponse { struct MessageResponse {
role: String, role: String,
content: String, content: String,
@@ -209,11 +208,12 @@ impl ChatClient {
Ok(r) => r, Ok(r) => r,
Err(e) => { Err(e) => {
last_error = Some(anyhow!("Request failed: {}", e)); last_error = Some(anyhow!("Request failed: {}", e));
if (e.is_timeout() || e.is_status()) if e.is_timeout() || e.is_status() {
&& attempt < self.max_retries - 1 { if attempt < self.max_retries - 1 {
tokio::time::sleep(Duration::from_millis(100 * 2_u64.pow(attempt))).await; tokio::time::sleep(Duration::from_millis(100 * 2_u64.pow(attempt))).await;
continue; continue;
} }
}
return Err(last_error.unwrap()); return Err(last_error.unwrap());
} }
}; };
+2 -4
View File
@@ -27,7 +27,6 @@ struct EmbeddingRequest {
} }
#[derive(Debug, Deserialize)] #[derive(Debug, Deserialize)]
#[allow(dead_code)]
#[serde(untagged)] #[serde(untagged)]
enum EmbeddingResponse { enum EmbeddingResponse {
Success { Success {
@@ -43,7 +42,6 @@ enum EmbeddingResponse {
} }
#[derive(Debug, Deserialize)] #[derive(Debug, Deserialize)]
#[allow(dead_code)]
struct EmbeddingData { struct EmbeddingData {
embedding: Vec<f32>, embedding: Vec<f32>,
#[serde(default)] #[serde(default)]
@@ -122,10 +120,10 @@ impl EmbeddingsClient {
/// Embed a single text string, returning a 768-dim vector /// Embed a single text string, returning a 768-dim vector
pub async fn embed_one(&self, text: &str) -> Result<Vector> { pub async fn embed_one(&self, text: &str) -> Result<Vector> {
let embeddings = self.embed(&[text.to_string()]).await?; let embeddings = self.embed(&[text.to_string()]).await?;
embeddings Ok(embeddings
.into_iter() .into_iter()
.next() .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 /// Embed multiple texts, batched at ≤32 per request, preserving input order
-358
View File
@@ -1,358 +0,0 @@
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
View File
@@ -1,6 +1,7 @@
use chrono::{DateTime, Utc}; use chrono::{DateTime, Utc};
use sqlx::PgPool; use sqlx::PgPool;
use uuid::Uuid; use uuid::Uuid;
use serde_json::json;
/// Minimal audit logger - records version snapshots on mutation /// Minimal audit logger - records version snapshots on mutation
#[derive(Clone)] #[derive(Clone)]
-1
View File
@@ -8,7 +8,6 @@ pub mod edge_repo;
pub mod community_repo; pub mod community_repo;
pub mod versioning; pub mod versioning;
pub mod audit_logger; pub mod audit_logger;
pub mod agent_repo;
// pub mod db_repo; // TODO: Fix Entity schema integration // pub mod db_repo; // TODO: Fix Entity schema integration
pub use event_log::{EventRecord, LogWriter}; pub use event_log::{EventRecord, LogWriter};
-131
View File
@@ -1,131 +0,0 @@
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"
-76
View File
@@ -1,76 +0,0 @@
---
# 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"]
-150
View File
@@ -1,150 +0,0 @@
---
# Tekton Task: Integration Tests for Poimen Memory Service
#
# Executes:
# 1. Database migrations
# 2. Integration test suites (cargo test)
# 3. Reports results
#
# Parameters:
# - image: Docker image with SHA to test
#
# Results:
# - summary: Test summary (pass/fail + count)
apiVersion: tekton.dev/v1
kind: Task
metadata:
name: poimen-integration-test
namespace: poimen
spec:
params:
- name: image
type: string
description: "Docker image SHA to test (e.g., forgejo.riotpiao.com/riotpiao-poimen/poimen-memory:abc123)"
results:
- name: summary
description: "Test summary: PASS or FAIL + test count"
steps:
# Step 1: Apply database migrations
- name: migrate
image: $(params.image)
env:
- name: DB_HOST
value: "memory-db-rw.poimen.svc.cluster.local"
- name: DB_PORT
value: "5432"
- name: DB_NAME
value: "memory"
- name: DB_USER
value: "app"
- name: DB_PASSWORD
valueFrom:
secretKeyRef:
name: memory-db-app
key: password
script: |
#!/bin/bash
set -e
echo "=========================================="
echo "Step 1: Database Migrations"
echo "=========================================="
echo ""
# Run migrations
/app/migrations/run_migrations.sh
echo ""
echo "✓ Migrations complete"
# Step 2: Run integration tests
- name: test
image: $(params.image)
env:
- name: DATABASE_URL
value: "postgresql://[email protected]:5432/memory"
- name: RUST_LOG
value: "info,mem_cli=debug,mem_ingest=debug,mem_store=debug"
- name: MEM_AUTH_MODE
value: "none"
- name: SQLX_OFFLINE
value: "true"
- name: PGPASSWORD
valueFrom:
secretKeyRef:
name: memory-db-app
key: password
script: |
#!/bin/bash
set -e
echo "=========================================="
echo "Step 2: Integration Tests"
echo "=========================================="
echo ""
TEST_SUITES=(
"it_phase3_phase4"
"it_unified_query_4_6"
"it_temporal_filtering_4_2_fixed"
)
PASSED=0
FAILED=0
for suite in "${TEST_SUITES[@]}"; do
echo "Running: $suite"
if cargo test --test "$suite" --lib 2>&1 | tail -50; then
((PASSED++))
echo "✓ $suite passed"
else
((FAILED++))
echo "✗ $suite failed"
fi
echo ""
done
# Run unit tests
echo "Running unit tests..."
if cargo test --lib mem_ingest 2>&1 | tail -100; then
echo "✓ mem_ingest passed"
else
((FAILED++))
echo "✗ mem_ingest failed"
fi
echo ""
if cargo test --lib mem_cli::query 2>&1 | tail -100; then
echo "✓ mem_cli::query passed"
else
((FAILED++))
echo "✗ mem_cli::query failed"
fi
echo ""
echo "=========================================="
echo "Test Summary: $PASSED passed, $FAILED failed"
echo "=========================================="
if [ $FAILED -eq 0 ]; then
echo "PASS: All integration tests passed"
echo "PASS: All integration tests passed" > /tekton/results/summary
exit 0
else
echo "FAIL: $FAILED test suite(s) failed"
echo "FAIL: $FAILED test suite(s) failed" > /tekton/results/summary
exit 1
fi
resources:
requests:
memory: "1Gi"
cpu: "500m"
limits:
memory: "2Gi"
cpu: "2000m"
-140
View File
@@ -1,140 +0,0 @@
---
# 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"
-142
View File
@@ -1,142 +0,0 @@
-- 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),
FOREIGN KEY (project_id) REFERENCES projects(id) ON DELETE CASCADE
);
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(),
UNIQUE(project_id, agent_id, DATE(recorded_at))
);
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);
-127
View File
@@ -1,127 +0,0 @@
#!/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
-608
View File
@@ -1,608 +0,0 @@
// 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");
}
}