From c1dd7aa4a3196a64163858bb93e8ee6c118de23d Mon Sep 17 00:00:00 2001 From: RCLL Date: Sun, 23 Aug 2026 23:50:02 +0300 Subject: [PATCH] feat(tunnels): cross-bank memory bridges (ADR-145 phase 4) Recovered from the 2026-06-27 snapshot import by classifying the base..snapshot delta at line granularity. Upstream base: d054b884 (2026-04-10). --- hindsight-api-slim/hindsight_api/api/http.py | 119 ++++++++++ .../hindsight_api/engine/memory_engine.py | 214 ++++++++++++++++++ hindsight-api-slim/hindsight_api/models.py | 33 +++ 3 files changed, 366 insertions(+) diff --git a/hindsight-api-slim/hindsight_api/api/http.py b/hindsight-api-slim/hindsight_api/api/http.py index bdb91329..0bb2379e 100644 --- a/hindsight-api-slim/hindsight_api/api/http.py +++ b/hindsight-api-slim/hindsight_api/api/http.py @@ -995,6 +995,48 @@ class BackgroundResponse(BaseModel): disposition: DispositionTraits | None = None +class CreateTunnelRequest(BaseModel): + """Request to create a cross-bank memory bridge.""" + + source_bank: str = Field(description="Source bank ID") + source_memory: str = Field(description="Source memory unit UUID") + target_bank: str = Field(description="Target bank ID") + target_memory: str = Field(description="Target memory unit UUID") + relation: Literal["same_concept", "depends_on", "contradicts", "extends"] = Field( + description="Relationship type between the memories" + ) + confidence: float = Field(default=0.8, ge=0.0, le=1.0, description="Confidence score (0.0-1.0)") + created_by: str | None = Field(default=None, description="Agent slug or user who creates the tunnel") + + +class TunnelItem(BaseModel): + """A single tunnel.""" + + id: str + source_bank: str + source_memory: str + target_bank: str + target_memory: str + relation: str + confidence: float + created_by: str | None = None + created_at: str + + +class CreateTunnelResponse(BaseModel): + """Response from tunnel creation.""" + + success: bool + tunnel: TunnelItem + + +class ListTunnelsResponse(BaseModel): + """Response from listing tunnels.""" + + tunnels: list[TunnelItem] = [] + total: int = 0 + + class DeleteTunnelResponse(BaseModel): """Response from deleting a tunnel.""" @@ -5982,6 +6024,83 @@ def _register_routes(app: FastAPI): logger.error(f"Error getting audit log stats: {traceback.format_exc()}") raise HTTPException(status_code=500, detail=str(e)) + + # ==================== Tunnel Endpoints (ADR-145 Phase 4) ==================== + + @app.post( + "/v1/default/banks/{bank_id}/tunnels", + response_model=CreateTunnelResponse, + summary="Create tunnel (cross-bank bridge)", + description="Create a cross-bank memory bridge between two memories in different banks. ADR-145 RCLL.", + operation_id="create_tunnel", + tags=["Memory"], + ) + async def api_create_tunnel( + bank_id: str, request: CreateTunnelRequest, request_context: RequestContext = Depends(get_request_context) + ): + """Create a tunnel between two memories.""" + try: + tunnel = await app.state.memory.create_tunnel_async( + source_bank=request.source_bank, + source_memory=request.source_memory, + target_bank=request.target_bank, + target_memory=request.target_memory, + relation=request.relation, + confidence=request.confidence, + created_by=request.created_by, + request_context=request_context, + ) + return CreateTunnelResponse(success=True, tunnel=TunnelItem(**tunnel)) + except ValueError as e: + raise HTTPException(status_code=400, detail=str(e)) + except (AuthenticationError, HTTPException): + raise + except Exception as e: + import traceback + + logger.error(f"Error creating tunnel for bank {bank_id}: {traceback.format_exc()}") + raise HTTPException(status_code=500, detail=str(e)) + + @app.get( + "/v1/default/banks/{bank_id}/tunnels", + response_model=ListTunnelsResponse, + summary="List tunnels", + description="List cross-bank memory bridges for a bank (both as source and target).", + operation_id="list_tunnels", + tags=["Memory"], + ) + async def api_list_tunnels( + bank_id: str, + relation: str | None = Query(default=None, description="Filter by relation type"), + target_bank: str | None = Query(default=None, description="Filter by target bank"), + request_context: RequestContext = Depends(get_request_context), + ): + """List tunnels for a bank.""" + try: + result = await app.state.memory.list_tunnels_async( + bank_id=bank_id, + relation=relation, + target_bank=target_bank, + request_context=request_context, + ) + tunnels = [TunnelItem(**t) for t in result["tunnels"]] + return ListTunnelsResponse(tunnels=tunnels, total=result["total"]) + except (AuthenticationError, HTTPException): + raise + except Exception as e: + import traceback + + logger.error(f"Error listing tunnels for bank {bank_id}: {traceback.format_exc()}") + raise HTTPException(status_code=500, detail=str(e)) + + @app.delete( + "/v1/default/banks/{bank_id}/tunnels/{tunnel_id}", + response_model=DeleteTunnelResponse, + summary="Delete tunnel", + description="Delete a cross-bank memory bridge.", + operation_id="delete_tunnel", + tags=["Memory"], + ) async def api_delete_tunnel( bank_id: str, tunnel_id: str, request_context: RequestContext = Depends(get_request_context) ): diff --git a/hindsight-api-slim/hindsight_api/engine/memory_engine.py b/hindsight-api-slim/hindsight_api/engine/memory_engine.py index e6ea715a..313690df 100644 --- a/hindsight-api-slim/hindsight_api/engine/memory_engine.py +++ b/hindsight-api-slim/hindsight_api/engine/memory_engine.py @@ -8166,6 +8166,220 @@ class MemoryEngine(MemoryEngineInterface): result_metadata={"mental_model_id": mental_model_id, "name": mental_model["name"]}, dedupe_by_bank=False, ) + + # ==================== Tunnel Methods (ADR-145 Phase 4) ==================== + + async def create_tunnel_async( + self, + source_bank: str, + source_memory: str, + target_bank: str, + target_memory: str, + relation: str, + confidence: float = 0.8, + created_by: str | None = None, + *, + request_context: "RequestContext", + ) -> dict[str, Any]: + """Create a cross-bank tunnel between two memories.""" + await self._authenticate_tenant(request_context) + pool = await self._get_pool() + + async with acquire_with_retry(pool) as conn: + # Validate source memory exists + source_row = await conn.fetchrow( + f"SELECT id FROM {fq_table('memory_units')} WHERE id = $1 AND bank_id = $2", + uuid.UUID(source_memory), + source_bank, + ) + if not source_row: + raise ValueError(f"Source memory {source_memory} not found in bank {source_bank}") + + # Validate target memory exists + target_row = await conn.fetchrow( + f"SELECT id FROM {fq_table('memory_units')} WHERE id = $1 AND bank_id = $2", + uuid.UUID(target_memory), + target_bank, + ) + if not target_row: + raise ValueError(f"Target memory {target_memory} not found in bank {target_bank}") + + # Validate relation type + valid_relations = ("same_concept", "depends_on", "contradicts", "extends") + if relation not in valid_relations: + raise ValueError(f"Invalid relation '{relation}'. Must be one of: {valid_relations}") + + # Insert tunnel (ON CONFLICT returns existing) + row = await conn.fetchrow( + f""" + INSERT INTO {fq_table('tunnels')} + (source_bank, source_memory, target_bank, target_memory, relation, confidence, created_by) + VALUES ($1, $2, $3, $4, $5, $6, $7) + ON CONFLICT (source_bank, source_memory, target_bank, target_memory, relation) DO UPDATE + SET confidence = EXCLUDED.confidence, created_by = EXCLUDED.created_by + RETURNING id, source_bank, source_memory, target_bank, target_memory, relation, confidence, created_by, created_at + """, + source_bank, + uuid.UUID(source_memory), + target_bank, + uuid.UUID(target_memory), + relation, + confidence, + created_by, + ) + + return { + "id": str(row["id"]), + "source_bank": row["source_bank"], + "source_memory": str(row["source_memory"]), + "target_bank": row["target_bank"], + "target_memory": str(row["target_memory"]), + "relation": row["relation"], + "confidence": row["confidence"], + "created_by": row["created_by"], + "created_at": row["created_at"].isoformat() if row["created_at"] else None, + } + + async def list_tunnels_async( + self, + bank_id: str, + relation: str | None = None, + target_bank: str | None = None, + *, + request_context: "RequestContext", + ) -> dict[str, Any]: + """List tunnels for a bank (as source OR target).""" + await self._authenticate_tenant(request_context) + pool = await self._get_pool() + + async with acquire_with_retry(pool) as conn: + where_clauses = ["(source_bank = $1 OR target_bank = $1)"] + params: list[Any] = [bank_id] + idx = 2 + + if relation: + where_clauses.append(f"relation = ${idx}") + params.append(relation) + idx += 1 + + if target_bank: + where_clauses.append(f"(source_bank = ${idx} OR target_bank = ${idx})") + params.append(target_bank) + idx += 1 + + where_sql = " AND ".join(where_clauses) + + rows = await conn.fetch( + f""" + SELECT id, source_bank, source_memory, target_bank, target_memory, + relation, confidence, created_by, created_at + FROM {fq_table('tunnels')} + WHERE {where_sql} + ORDER BY created_at DESC + """, + *params, + ) + + tunnels = [ + { + "id": str(row["id"]), + "source_bank": row["source_bank"], + "source_memory": str(row["source_memory"]), + "target_bank": row["target_bank"], + "target_memory": str(row["target_memory"]), + "relation": row["relation"], + "confidence": row["confidence"], + "created_by": row["created_by"], + "created_at": row["created_at"].isoformat() if row["created_at"] else None, + } + for row in rows + ] + + return {"tunnels": tunnels, "total": len(tunnels)} + + async def delete_tunnel_async( + self, + bank_id: str, + tunnel_id: str, + *, + request_context: "RequestContext", + ) -> dict[str, Any]: + """Delete a tunnel.""" + await self._authenticate_tenant(request_context) + pool = await self._get_pool() + + async with acquire_with_retry(pool) as conn: + result = await conn.execute( + f""" + DELETE FROM {fq_table('tunnels')} + WHERE id = $1 AND (source_bank = $2 OR target_bank = $2) + """, + uuid.UUID(tunnel_id), + bank_id, + ) + deleted = result.split()[-1] != "0" # "DELETE N" + return {"success": True, "deleted": deleted} + + async def get_tunneled_memories_async( + self, + bank_id: str, + memory_ids: list[str], + *, + request_context: "RequestContext", + ) -> list[dict[str, Any]]: + """Get related memories from other banks via tunnels. + + Used during recall to fetch cross-bank context. + For each memory_id in the recall results, finds tunnels and fetches the other side's memory text. + """ + if not memory_ids: + return [] + + await self._authenticate_tenant(request_context) + pool = await self._get_pool() + + uuid_ids = [uuid.UUID(mid) for mid in memory_ids] + + async with acquire_with_retry(pool) as conn: + # Find tunnels where our bank+memories are either source or target + rows = await conn.fetch( + f""" + SELECT t.id AS tunnel_id, t.source_bank, t.source_memory, t.target_bank, t.target_memory, + t.relation, t.confidence, + mu.text AS linked_text, mu.bank_id AS linked_bank + FROM {fq_table('tunnels')} t + LEFT JOIN {fq_table('memory_units')} mu ON ( + CASE + WHEN t.source_bank = $1 AND t.source_memory = ANY($2::uuid[]) + THEN mu.id = t.target_memory AND mu.bank_id = t.target_bank + WHEN t.target_bank = $1 AND t.target_memory = ANY($2::uuid[]) + THEN mu.id = t.source_memory AND mu.bank_id = t.source_bank + ELSE FALSE + END + ) + WHERE (t.source_bank = $1 AND t.source_memory = ANY($2::uuid[])) + OR (t.target_bank = $1 AND t.target_memory = ANY($2::uuid[])) + """, + bank_id, + uuid_ids, + ) + + results = [] + for row in rows: + results.append({ + "tunnel_id": str(row["tunnel_id"]), + "relation": row["relation"], + "confidence": row["confidence"], + "linked_bank": row["linked_bank"], + "linked_text": row["linked_text"], + "source_bank": row["source_bank"], + "source_memory": str(row["source_memory"]), + "target_bank": row["target_bank"], + "target_memory": str(row["target_memory"]), + }) + + return results + # ==================== Closet Methods (ADR-145 Phase 3) ==================== async def create_closets_async( diff --git a/hindsight-api-slim/hindsight_api/models.py b/hindsight-api-slim/hindsight_api/models.py index 74ef412f..d3a7c9a8 100644 --- a/hindsight-api-slim/hindsight_api/models.py +++ b/hindsight-api-slim/hindsight_api/models.py @@ -296,6 +296,39 @@ class Bank(Base): updated_at: Mapped[datetime] = mapped_column(TIMESTAMP(timezone=True), server_default=func.now()) __table_args__ = (Index("idx_banks_bank_id", "bank_id"),) +class Tunnel(Base): + """Cross-bank memory bridges — links between concepts in different banks (ADR-145 RCLL).""" + + __tablename__ = "tunnels" + + id: Mapped[PyUUID] = mapped_column( + UUID(as_uuid=True), primary_key=True, server_default=sql_text("gen_random_uuid()") + ) + source_bank: Mapped[str] = mapped_column(Text, nullable=False) + source_memory: Mapped[PyUUID] = mapped_column(UUID(as_uuid=True), nullable=False) + target_bank: Mapped[str] = mapped_column(Text, nullable=False) + target_memory: Mapped[PyUUID] = mapped_column(UUID(as_uuid=True), nullable=False) + relation: Mapped[str] = mapped_column(Text, nullable=False) # same_concept | depends_on | contradicts | extends + confidence: Mapped[float] = mapped_column(Float, nullable=False, server_default="0.8") + created_by: Mapped[str | None] = mapped_column(Text) # agent slug or user who created + metadata_: Mapped[dict] = mapped_column("metadata", JSONB, server_default=sql_text("'{}'::jsonb")) + created_at: Mapped[datetime] = mapped_column(TIMESTAMP(timezone=True), server_default=func.now()) + + __table_args__ = ( + CheckConstraint( + "relation IN ('same_concept', 'depends_on', 'contradicts', 'extends')", + name="tunnels_relation_check", + ), + CheckConstraint("confidence >= 0.0 AND confidence <= 1.0", name="tunnels_confidence_check"), + CheckConstraint("source_bank != target_bank OR source_memory != target_memory", name="tunnels_no_self_loop"), + Index("idx_tunnels_source", "source_bank", "source_memory"), + Index("idx_tunnels_target", "target_bank", "target_memory"), + Index("idx_tunnels_source_bank", "source_bank"), + Index("idx_tunnels_target_bank", "target_bank"), + Index("idx_tunnels_relation", "relation"), + ) + + class Closet(Base): """Compressed memory summaries with pointers to source facts (ADR-145 RCLL)."""