From 1dd73787324deec603169bffc40b60cbcac61b25 Mon Sep 17 00:00:00 2001 From: RCLL Date: Sun, 23 Aug 2026 23:50:02 +0300 Subject: [PATCH] feat(closets): compress memories by room+hall into closets (ADR-145 ph.3) 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 | 131 ++++++++++ .../hindsight_api/engine/memory_engine.py | 235 ++++++++++++++++++ hindsight-api-slim/hindsight_api/models.py | 29 +++ 3 files changed, 395 insertions(+) diff --git a/hindsight-api-slim/hindsight_api/api/http.py b/hindsight-api-slim/hindsight_api/api/http.py index 321c0ffc..bdb91329 100644 --- a/hindsight-api-slim/hindsight_api/api/http.py +++ b/hindsight-api-slim/hindsight_api/api/http.py @@ -995,6 +995,52 @@ class BackgroundResponse(BaseModel): disposition: DispositionTraits | None = None +class DeleteTunnelResponse(BaseModel): + """Response from deleting a tunnel.""" + + success: bool + deleted: bool = False + + +# ── Closet models (ADR-145 Phase 3) ────────────────────────── + + +class CreateClosetRequest(BaseModel): + """Request model for creating a closet (compressed summary).""" + + room: str | None = Field(default=None, description="Room (topic) to compress. If not provided, compresses all.") + hall: str | None = Field(default=None, description="Hall (knowledge type) to compress. If not provided, compresses all.") + min_sources: int = Field(default=5, description="Minimum number of source memories to create a closet (default: 5)") + query: str | None = Field(default=None, description="Optional query to guide compression focus") + + +class ClosetItem(BaseModel): + """A single closet.""" + + id: str + summary: str + source_count: int + room: str | None = None + hall: str | None = None + token_count: int = 0 + created_at: str + + +class CreateClosetResponse(BaseModel): + """Response from closet creation.""" + + success: bool + closets_created: int = 0 + closets: list[ClosetItem] = FieldWithDefault(list, description="Created closets") + + +class ListClosetsResponse(BaseModel): + """Response from listing closets.""" + + closets: list[ClosetItem] = FieldWithDefault(list, description="Closets") + total: int = 0 + + class BankListItem(BaseModel): """Bank list item with profile summary.""" @@ -5936,3 +5982,88 @@ def _register_routes(app: FastAPI): logger.error(f"Error getting audit log stats: {traceback.format_exc()}") raise HTTPException(status_code=500, detail=str(e)) + async def api_delete_tunnel( + bank_id: str, tunnel_id: str, request_context: RequestContext = Depends(get_request_context) + ): + """Delete a tunnel.""" + try: + result = await app.state.memory.delete_tunnel_async( + bank_id=bank_id, + tunnel_id=tunnel_id, + request_context=request_context, + ) + return DeleteTunnelResponse(**result) + except (AuthenticationError, HTTPException): + raise + except Exception as e: + import traceback + + logger.error(f"Error deleting tunnel {tunnel_id}: {traceback.format_exc()}") + raise HTTPException(status_code=500, detail=str(e)) + + # ── Closet endpoints (ADR-145 Phase 3) ──────────────────────── + + @app.post( + "/v1/default/banks/{bank_id}/closets", + response_model=CreateClosetResponse, + summary="Create closets (compressed summaries)", + description="Compress memories by room+hall into closets. ADR-145 RCLL Phase 3.", + operation_id="create_closets", + tags=["Memory"], + ) + async def api_create_closets( + bank_id: str, request: CreateClosetRequest, request_context: RequestContext = Depends(get_request_context) + ): + """Create compressed memory closets.""" + try: + result = await app.state.memory.create_closets_async( + bank_id=bank_id, + room=request.room, + hall=request.hall, + min_sources=request.min_sources, + query=request.query, + request_context=request_context, + ) + return CreateClosetResponse(**result) + except OperationValidationError as e: + raise HTTPException(status_code=e.status_code, detail=e.reason) + except (AuthenticationError, HTTPException): + raise + except Exception as e: + import traceback + + logger.error(f"Error creating closets for bank {bank_id}: {traceback.format_exc()}") + raise HTTPException(status_code=500, detail=str(e)) + + @app.get( + "/v1/default/banks/{bank_id}/closets", + response_model=ListClosetsResponse, + summary="List closets", + description="List all compressed memory summaries for a bank.", + operation_id="list_closets", + tags=["Memory"], + ) + async def api_list_closets( + bank_id: str, + room: str | None = Query(None, description="Filter by room"), + hall: str | None = Query(None, description="Filter by hall"), + request_context: RequestContext = Depends(get_request_context), + ): + """List closets for a bank.""" + try: + result = await app.state.memory.list_closets_async( + bank_id=bank_id, + room=room, + hall=hall, + request_context=request_context, + ) + return ListClosetsResponse(**result) + except OperationValidationError as e: + raise HTTPException(status_code=e.status_code, detail=e.reason) + except (AuthenticationError, HTTPException): + raise + except Exception as e: + import traceback + + logger.error(f"Error listing closets for bank {bank_id}: {traceback.format_exc()}") + raise HTTPException(status_code=500, detail=str(e)) diff --git a/hindsight-api-slim/hindsight_api/engine/memory_engine.py b/hindsight-api-slim/hindsight_api/engine/memory_engine.py index fe37e27d..e6ea715a 100644 --- a/hindsight-api-slim/hindsight_api/engine/memory_engine.py +++ b/hindsight-api-slim/hindsight_api/engine/memory_engine.py @@ -89,6 +89,8 @@ _PROTECTED_TABLES = frozenset( "chunks", "async_operations", "file_storage", + "tunnels", + "closets", ] ) @@ -8164,3 +8166,236 @@ class MemoryEngine(MemoryEngineInterface): result_metadata={"mental_model_id": mental_model_id, "name": mental_model["name"]}, dedupe_by_bank=False, ) + # ==================== Closet Methods (ADR-145 Phase 3) ==================== + + async def create_closets_async( + self, + bank_id: str, + room: str | None = None, + hall: str | None = None, + min_sources: int = 5, + query: str | None = None, + request_context: "RequestContext | None" = None, + ) -> dict[str, Any]: + """ + Create compressed closets from memories grouped by room+hall. + + Flow: + 1. Query memory_units grouped by (room, hall) with count >= min_sources + 2. For each group, if no existing closet: use LLM to summarize + 3. Generate embedding for the summary + 4. Store as Closet with source_ids + """ + from .retain import embedding_utils + + await self._authenticate_tenant(request_context) + pool = await self._get_pool() + + closets_created = [] + + async with acquire_with_retry(pool) as conn: + # Build WHERE clause for optional room/hall filters + where_parts = ["bank_id = $1"] + params: list[Any] = [bank_id] + idx = 2 + + if room: + where_parts.append(f"room = ${idx}") + params.append(room) + idx += 1 + if hall: + where_parts.append(f"hall = ${idx}") + params.append(hall) + idx += 1 + + where_sql = " AND ".join(where_parts) + + # Find groups with enough memories to compress + groups = await conn.fetch( + f""" + SELECT room, hall, array_agg(id) AS ids, count(*) AS cnt + FROM {fq_table("memory_units")} + WHERE {where_sql} AND room IS NOT NULL + GROUP BY room, hall + HAVING count(*) >= ${idx} + ORDER BY count(*) DESC + """, + *params, + min_sources, + ) + + for group in groups: + g_room = group["room"] + g_hall = group["hall"] + source_ids = [str(uid) for uid in group["ids"]] + + # Check if closet already exists for this room+hall + existing = await conn.fetchval( + f""" + SELECT id FROM {fq_table("closets")} + WHERE bank_id = $1 AND room IS NOT DISTINCT FROM $2 AND hall IS NOT DISTINCT FROM $3 + LIMIT 1 + """, + bank_id, + g_room, + g_hall, + ) + if existing: + continue + + # Collect source memory texts + mem_rows = await conn.fetch( + f""" + SELECT text FROM {fq_table("memory_units")} + WHERE bank_id = $1 AND id = ANY($2::uuid[]) + ORDER BY event_date DESC + LIMIT 100 + """, + bank_id, + group["ids"], + ) + source_texts = [r["text"] for r in mem_rows] + combined = "\n".join(f"- {t}" for t in source_texts) + + # Build LLM prompt for compression + focus = f" Focus especially on: {query}" if query else "" + messages = [ + { + "role": "system", + "content": ( + "You are a memory compression assistant. " + "Summarize the following facts into a dense, information-rich paragraph. " + "Preserve key details, names, dates, and decisions. " + "Do NOT add opinions or speculation — only compress what is stated." + ), + }, + { + "role": "user", + "content": ( + f"Room (topic): {g_room or 'general'}\n" + f"Hall (type): {g_hall or 'mixed'}\n" + f"Number of facts: {len(source_texts)}\n" + f"{focus}\n\n" + f"Facts to compress:\n{combined}" + ), + }, + ] + + try: + summary = await self._llm_config.call( + messages=messages, + max_completion_tokens=1024, + temperature=0.2, + scope="closet_compress", + ) + except Exception as e: + logger.error(f"[CLOSET] LLM error for room={g_room}, hall={g_hall}: {e}") + continue + + if not summary or not isinstance(summary, str): + continue + + # Count tokens + token_cnt = count_tokens(summary) + + # Generate embedding + try: + emb = await embedding_utils.generate_embeddings_batch(self.embeddings, [summary]) + embedding_str = str(emb[0]) if emb else None + except Exception as e: + logger.error(f"[CLOSET] Embedding error for room={g_room}, hall={g_hall}: {e}") + embedding_str = None + + # Insert closet + row = await conn.fetchrow( + f""" + INSERT INTO {fq_table("closets")} + (bank_id, summary, source_ids, room, hall, token_count, embedding) + VALUES ($1, $2, $3::jsonb, $4, $5, $6, $7) + RETURNING id, created_at + """, + bank_id, + summary, + json.dumps(source_ids), + g_room, + g_hall, + token_cnt, + embedding_str, + ) + + closets_created.append({ + "id": str(row["id"]), + "summary": summary, + "source_count": len(source_ids), + "room": g_room, + "hall": g_hall, + "token_count": token_cnt, + "created_at": row["created_at"].isoformat(), + }) + + logger.info( + f"[CLOSET] Created closet for bank={bank_id} room={g_room} hall={g_hall} " + f"sources={len(source_ids)} tokens={token_cnt}" + ) + + return { + "success": True, + "closets_created": len(closets_created), + "closets": closets_created, + } + + async def list_closets_async( + self, + bank_id: str, + room: str | None = None, + hall: str | None = None, + request_context: "RequestContext | None" = None, + ) -> dict[str, Any]: + """List closets for a bank, optionally filtered by room/hall.""" + await self._authenticate_tenant(request_context) + pool = await self._get_pool() + + async with acquire_with_retry(pool) as conn: + where_parts = ["bank_id = $1"] + params: list[Any] = [bank_id] + idx = 2 + + if room: + where_parts.append(f"room = ${idx}") + params.append(room) + idx += 1 + if hall: + where_parts.append(f"hall = ${idx}") + params.append(hall) + idx += 1 + + where_sql = " AND ".join(where_parts) + + rows = await conn.fetch( + f""" + SELECT id, summary, source_ids, room, hall, token_count, created_at + FROM {fq_table("closets")} + WHERE {where_sql} + ORDER BY created_at DESC + """, + *params, + ) + + closets = [] + for row in rows: + raw_ids = row["source_ids"] if row["source_ids"] else [] + src_ids = json.loads(raw_ids) if isinstance(raw_ids, str) else raw_ids + closets.append({ + "id": str(row["id"]), + "summary": row["summary"], + "source_count": len(src_ids) if isinstance(src_ids, list) else 0, + "room": row["room"], + "hall": row["hall"], + "token_count": row["token_count"], + "created_at": row["created_at"].isoformat(), + }) + + return { + "closets": closets, + "total": len(closets), + } diff --git a/hindsight-api-slim/hindsight_api/models.py b/hindsight-api-slim/hindsight_api/models.py index bd854dcf..74ef412f 100644 --- a/hindsight-api-slim/hindsight_api/models.py +++ b/hindsight-api-slim/hindsight_api/models.py @@ -296,3 +296,32 @@ 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 Closet(Base): + """Compressed memory summaries with pointers to source facts (ADR-145 RCLL).""" + + __tablename__ = "closets" + + id: Mapped[PyUUID] = mapped_column( + UUID(as_uuid=True), primary_key=True, server_default=sql_text("gen_random_uuid()") + ) + bank_id: Mapped[str] = mapped_column(Text, nullable=False) + summary: Mapped[str] = mapped_column(Text, nullable=False) + source_ids: Mapped[list] = mapped_column(JSONB, server_default=sql_text("'[]'::jsonb")) # UUIDs of source memory_units + room: Mapped[str | None] = mapped_column(Text) + hall: Mapped[str | None] = mapped_column(Text) + token_count: Mapped[int] = mapped_column(Integer, server_default="0") + embedding = mapped_column(Vector(EMBEDDING_DIMENSION)) # pgvector for recall search + created_at: Mapped[datetime] = mapped_column(TIMESTAMP(timezone=True), server_default=func.now()) + updated_at: Mapped[datetime] = mapped_column(TIMESTAMP(timezone=True), server_default=func.now()) + + __table_args__ = ( + Index("idx_closets_bank_id", "bank_id"), + Index("idx_closets_bank_room", "bank_id", "room"), + Index("idx_closets_bank_room_hall", "bank_id", "room", "hall"), + Index( + "idx_closets_embedding", + "embedding", + postgresql_using="hnsw", + postgresql_ops={"embedding": "vector_cosine_ops"}, + ), + )