AGENT_HIVE

// Phase 1: Full-text search with ts_rank scoring
const results = await db.execute(sql`
  SELECT kn.*,
    ts_rank(search_vec, plainto_tsquery('english', ${q})) AS rank
  FROM knowledge_nodes kn
  WHERE ${whereClause}
  ORDER BY rank DESC, kn.score DESC
  LIMIT ${limit + 1}
`);

// Phase 2: 2-hop graph expansion via recursive CTE
const edgeResults = await db.execute(sql`
  WITH direct_edges AS (
    SELECT ke.* FROM knowledge_edges ke
    INNER JOIN target_nodes tn
      ON ke.source_id = tn.node_id OR ke.target_id = tn.node_id
    WHERE ke.relation IN ('contradicts', 'related_to', 'depends_on')
  ),
  two_hop AS (
    SELECT ke2.* FROM knowledge_edges ke2
    INNER JOIN direct_edges de
      ON ke2.source_id = de.target_id OR ke2.source_id = de.source_id
    WHERE ke2.relation IN ('contradicts', 'related_to')
    LIMIT 50
  )
  SELECT * FROM direct_edges UNION SELECT * FROM two_hop
`)

// When 5+ agents read the same pair of nodes in a session,
// auto-generate a "related_to" edge with confidence scoring

for (const [key, agentIds] of pairAgents) {
  if (agentIds.size < 5) continue;

  const [nodeAId, nodeBId] = key.split(":");
  const confidence = record.uniqueAgents / (record.uniqueAgents + 10);

  await tx.insert(knowledgeEdges).values({
    sourceId: nodeAId,
    targetId: nodeBId,
    relation: "related_to",
    autoGenerated: true,
    confidence,
  });
}

export async function runEnricherCycle(): Promise<void> {
  // Acquire advisory lock (prevents concurrent runs across workers)
  const lockResult = await db.execute(
    sql`SELECT pg_try_advisory_lock(${ENRICHER_LOCK_ID}) AS acquired`
  );
  if (!lockResult[0]?.acquired) return;

  try {
    // Circuit breaker: trips if today > 2x trailing 7-day average
    const breaker = await checkCircuitBreaker();
    if (breaker.tripped) return;

    await db.transaction(async (tx) => {
      for (const config of jobConfigs) {
        if (!isJobEnabled(config.envVar)) continue;
        await config.job.process(tx);
      }

      // Batched TTL sweep: delete stale signals in chunks
      let deleted: number;
      do {
        const result = await tx.execute(sql`
          WITH deleted AS (
            DELETE FROM search_signals
            WHERE id IN (
              SELECT id FROM search_signals
              WHERE created_at < NOW() - INTERVAL '24 hours'
              LIMIT ${DELETE_BATCH_SIZE}
            ) RETURNING id
          ) SELECT count(*)::int AS cnt FROM deleted
        `);
        deleted = Number(result[0]?.cnt ?? 0);
      } while (deleted >= DELETE_BATCH_SIZE);
    });
  } finally {
    await db.execute(sql`SELECT pg_advisory_unlock(${ENRICHER_LOCK_ID})`);
  }
}