diff --git a/hindsight-api/hindsight_api/api/http.py b/hindsight-api/hindsight_api/api/http.py index 91e7a9aa..1af29103 100644 --- a/hindsight-api/hindsight_api/api/http.py +++ b/hindsight-api/hindsight_api/api/http.py @@ -1943,6 +1943,8 @@ def _register_routes(app: FastAPI): bank_id, type, limit=limit, q=q, tags=tags, tags_match=tags_match, request_context=request_context ) return data + except OperationValidationError as e: + raise HTTPException(status_code=e.status_code, detail=e.reason) except (AuthenticationError, HTTPException): raise except Exception as e: @@ -1991,6 +1993,8 @@ def _register_routes(app: FastAPI): request_context=request_context, ) return data + except OperationValidationError as e: + raise HTTPException(status_code=e.status_code, detail=e.reason) except (AuthenticationError, HTTPException): raise except Exception as e: @@ -2022,6 +2026,8 @@ def _register_routes(app: FastAPI): if data is None: raise HTTPException(status_code=404, detail=f"Memory unit '{memory_id}' not found") return data + except OperationValidationError as e: + raise HTTPException(status_code=e.status_code, detail=e.reason) except (AuthenticationError, HTTPException): raise except Exception as e: @@ -2365,6 +2371,13 @@ def _register_routes(app: FastAPI): try: # Authenticate and set tenant schema await app.state.memory._authenticate_tenant(request_context) + if app.state.memory._operation_validator: + from hindsight_api.extensions import BankReadContext + + ctx = BankReadContext(bank_id=bank_id, operation="get_bank_stats", request_context=request_context) + await app.state.memory._validate_operation( + app.state.memory._operation_validator.validate_bank_read(ctx) + ) pool = await app.state.memory._get_pool() async with acquire_with_retry(pool) as conn: # Get node counts by fact_type @@ -2498,6 +2511,8 @@ def _register_routes(app: FastAPI): total_observations=total_observations, ) + except OperationValidationError as e: + raise HTTPException(status_code=e.status_code, detail=e.reason) except (AuthenticationError, HTTPException): raise except Exception as e: @@ -2532,6 +2547,8 @@ def _register_routes(app: FastAPI): limit=data["limit"], offset=data["offset"], ) + except OperationValidationError as e: + raise HTTPException(status_code=e.status_code, detail=e.reason) except (AuthenticationError, HTTPException): raise except Exception as e: @@ -2571,6 +2588,8 @@ def _register_routes(app: FastAPI): for obs in entity["observations"] ], ) + except OperationValidationError as e: + raise HTTPException(status_code=e.status_code, detail=e.reason) except (AuthenticationError, HTTPException): raise except Exception as e: @@ -2632,6 +2651,8 @@ def _register_routes(app: FastAPI): request_context=request_context, ) return MentalModelListResponse(items=[MentalModelResponse(**m) for m in mental_models]) + except OperationValidationError as e: + raise HTTPException(status_code=e.status_code, detail=e.reason) except (AuthenticationError, HTTPException): raise except Exception as e: @@ -2790,6 +2811,8 @@ def _register_routes(app: FastAPI): if mental_model is None: raise HTTPException(status_code=404, detail=f"Mental model '{mental_model_id}' not found") return MentalModelResponse(**mental_model) + except OperationValidationError as e: + raise HTTPException(status_code=e.status_code, detail=e.reason) except (AuthenticationError, HTTPException): raise except Exception as e: @@ -2821,6 +2844,8 @@ def _register_routes(app: FastAPI): if not deleted: raise HTTPException(status_code=404, detail=f"Mental model '{mental_model_id}' not found") return {"status": "deleted"} + except OperationValidationError as e: + raise HTTPException(status_code=e.status_code, detail=e.reason) except (AuthenticationError, HTTPException): raise except Exception as e: @@ -2863,6 +2888,8 @@ def _register_routes(app: FastAPI): request_context=request_context, ) return DirectiveListResponse(items=[DirectiveResponse(**d) for d in directives]) + except OperationValidationError as e: + raise HTTPException(status_code=e.status_code, detail=e.reason) except (AuthenticationError, HTTPException): raise except Exception as e: @@ -2895,6 +2922,8 @@ def _register_routes(app: FastAPI): if directive is None: raise HTTPException(status_code=404, detail=f"Directive '{directive_id}' not found") return DirectiveResponse(**directive) + except OperationValidationError as e: + raise HTTPException(status_code=e.status_code, detail=e.reason) except (AuthenticationError, HTTPException): raise except Exception as e: @@ -2931,6 +2960,8 @@ def _register_routes(app: FastAPI): return DirectiveResponse(**directive) except ValueError as e: raise HTTPException(status_code=400, detail=str(e)) + except OperationValidationError as e: + raise HTTPException(status_code=e.status_code, detail=e.reason) except (AuthenticationError, HTTPException): raise except Exception as e: @@ -2969,6 +3000,8 @@ def _register_routes(app: FastAPI): if directive is None: raise HTTPException(status_code=404, detail=f"Directive '{directive_id}' not found") return DirectiveResponse(**directive) + except OperationValidationError as e: + raise HTTPException(status_code=e.status_code, detail=e.reason) except (AuthenticationError, HTTPException): raise except Exception as e: @@ -3000,6 +3033,8 @@ def _register_routes(app: FastAPI): if not deleted: raise HTTPException(status_code=404, detail=f"Directive '{directive_id}' not found") return {"status": "deleted"} + except OperationValidationError as e: + raise HTTPException(status_code=e.status_code, detail=e.reason) except (AuthenticationError, HTTPException): raise except Exception as e: @@ -3038,6 +3073,8 @@ def _register_routes(app: FastAPI): bank_id=bank_id, search_query=q, limit=limit, offset=offset, request_context=request_context ) return data + except OperationValidationError as e: + raise HTTPException(status_code=e.status_code, detail=e.reason) except (AuthenticationError, HTTPException): raise except Exception as e: @@ -3070,6 +3107,8 @@ def _register_routes(app: FastAPI): if not document: raise HTTPException(status_code=404, detail="Document not found") return document + except OperationValidationError as e: + raise HTTPException(status_code=e.status_code, detail=e.reason) except (AuthenticationError, HTTPException): raise except Exception as e: @@ -3123,6 +3162,8 @@ def _register_routes(app: FastAPI): request_context=request_context, ) return data + except OperationValidationError as e: + raise HTTPException(status_code=e.status_code, detail=e.reason) except (AuthenticationError, HTTPException): raise except Exception as e: @@ -3152,6 +3193,8 @@ def _register_routes(app: FastAPI): if not chunk: raise HTTPException(status_code=404, detail="Chunk not found") return chunk + except OperationValidationError as e: + raise HTTPException(status_code=e.status_code, detail=e.reason) except (AuthenticationError, HTTPException): raise except Exception as e: @@ -3196,6 +3239,8 @@ def _register_routes(app: FastAPI): document_id=document_id, memory_units_deleted=result["memory_units_deleted"], ) + except OperationValidationError as e: + raise HTTPException(status_code=e.status_code, detail=e.reason) except (AuthenticationError, HTTPException): raise except Exception as e: @@ -3232,6 +3277,8 @@ def _register_routes(app: FastAPI): offset=offset, operations=[OperationResponse(**op) for op in result["operations"]], ) + except OperationValidationError as e: + raise HTTPException(status_code=e.status_code, detail=e.reason) except (AuthenticationError, HTTPException): raise except Exception as e: @@ -3263,6 +3310,8 @@ def _register_routes(app: FastAPI): result = await app.state.memory.get_operation_status(bank_id, operation_id, request_context=request_context) return OperationStatusResponse(**result) + except OperationValidationError as e: + raise HTTPException(status_code=e.status_code, detail=e.reason) except (AuthenticationError, HTTPException): raise except Exception as e: @@ -3295,6 +3344,8 @@ def _register_routes(app: FastAPI): return CancelOperationResponse(**result) except ValueError as e: raise HTTPException(status_code=404, detail=str(e)) + except OperationValidationError as e: + raise HTTPException(status_code=e.status_code, detail=e.reason) except (AuthenticationError, HTTPException): raise except Exception as e: @@ -3331,6 +3382,8 @@ def _register_routes(app: FastAPI): mission=mission, background=mission, # Backwards compat ) + except OperationValidationError as e: + raise HTTPException(status_code=e.status_code, detail=e.reason) except (AuthenticationError, HTTPException): raise except Exception as e: @@ -3374,6 +3427,8 @@ def _register_routes(app: FastAPI): mission=mission, background=mission, # Backwards compat ) + except OperationValidationError as e: + raise HTTPException(status_code=e.status_code, detail=e.reason) except (AuthenticationError, HTTPException): raise except Exception as e: @@ -3402,6 +3457,8 @@ def _register_routes(app: FastAPI): ) mission = result.get("mission") or "" return BackgroundResponse(mission=mission, background=mission) + except OperationValidationError as e: + raise HTTPException(status_code=e.status_code, detail=e.reason) except (AuthenticationError, HTTPException): raise except Exception as e: @@ -3455,6 +3512,8 @@ def _register_routes(app: FastAPI): mission=mission, background=mission, # Backwards compat ) + except OperationValidationError as e: + raise HTTPException(status_code=e.status_code, detail=e.reason) except (AuthenticationError, HTTPException): raise except Exception as e: @@ -3508,6 +3567,8 @@ def _register_routes(app: FastAPI): mission=mission, background=mission, # Backwards compat ) + except OperationValidationError as e: + raise HTTPException(status_code=e.status_code, detail=e.reason) except (AuthenticationError, HTTPException): raise except Exception as e: @@ -3537,6 +3598,8 @@ def _register_routes(app: FastAPI): + result.get("entities_deleted", 0) + result.get("documents_deleted", 0), ) + except OperationValidationError as e: + raise HTTPException(status_code=e.status_code, detail=e.reason) except (AuthenticationError, HTTPException): raise except Exception as e: @@ -3563,6 +3626,8 @@ def _register_routes(app: FastAPI): message=f"Cleared {result.get('deleted_count', 0)} observations", deleted_count=result.get("deleted_count", 0), ) + except OperationValidationError as e: + raise HTTPException(status_code=e.status_code, detail=e.reason) except (AuthenticationError, HTTPException): raise except Exception as e: @@ -3595,6 +3660,8 @@ def _register_routes(app: FastAPI): request_context=request_context, ) return ClearMemoryObservationsResponse(deleted_count=result["deleted_count"]) + except OperationValidationError as e: + raise HTTPException(status_code=e.status_code, detail=e.reason) except (AuthenticationError, HTTPException): raise except Exception as e: @@ -3625,6 +3692,13 @@ def _register_routes(app: FastAPI): try: # Authenticate and set schema context for multi-tenant DB queries await app.state.memory._authenticate_tenant(request_context) + if app.state.memory._operation_validator: + from hindsight_api.extensions import BankReadContext + + ctx = BankReadContext(bank_id=bank_id, operation="get_bank_config", request_context=request_context) + await app.state.memory._validate_operation( + app.state.memory._operation_validator.validate_bank_read(ctx) + ) # Get resolved config from config resolver config_dict = await app.state.memory._config_resolver.get_bank_config(bank_id, request_context) @@ -3633,6 +3707,8 @@ def _register_routes(app: FastAPI): bank_overrides = await app.state.memory._config_resolver._load_bank_config(bank_id) return BankConfigResponse(bank_id=bank_id, config=config_dict, overrides=bank_overrides) + except OperationValidationError as e: + raise HTTPException(status_code=e.status_code, detail=e.reason) except (AuthenticationError, HTTPException): raise except Exception as e: @@ -3663,6 +3739,13 @@ def _register_routes(app: FastAPI): try: # Authenticate and set schema context for multi-tenant DB queries await app.state.memory._authenticate_tenant(request_context) + if app.state.memory._operation_validator: + from hindsight_api.extensions import BankWriteContext + + ctx = BankWriteContext(bank_id=bank_id, operation="update_bank_config", request_context=request_context) + await app.state.memory._validate_operation( + app.state.memory._operation_validator.validate_bank_write(ctx) + ) # Update config via config resolver (validates configurable fields and permissions) await app.state.memory._config_resolver.update_bank_config(bank_id, request.updates, request_context) @@ -3672,6 +3755,8 @@ def _register_routes(app: FastAPI): bank_overrides = await app.state.memory._config_resolver._load_bank_config(bank_id) return BankConfigResponse(bank_id=bank_id, config=config_dict, overrides=bank_overrides) + except OperationValidationError as e: + raise HTTPException(status_code=e.status_code, detail=e.reason) except ValueError as e: # Validation error (e.g., trying to override static field) raise HTTPException(status_code=400, detail=str(e)) @@ -3703,6 +3788,13 @@ def _register_routes(app: FastAPI): try: # Authenticate and set schema context for multi-tenant DB queries await app.state.memory._authenticate_tenant(request_context) + if app.state.memory._operation_validator: + from hindsight_api.extensions import BankWriteContext + + ctx = BankWriteContext(bank_id=bank_id, operation="reset_bank_config", request_context=request_context) + await app.state.memory._validate_operation( + app.state.memory._operation_validator.validate_bank_write(ctx) + ) # Reset config via config resolver await app.state.memory._config_resolver.reset_bank_config(bank_id) @@ -3712,6 +3804,8 @@ def _register_routes(app: FastAPI): bank_overrides = await app.state.memory._config_resolver._load_bank_config(bank_id) return BankConfigResponse(bank_id=bank_id, config=config_dict, overrides=bank_overrides) + except OperationValidationError as e: + raise HTTPException(status_code=e.status_code, detail=e.reason) except (AuthenticationError, HTTPException): raise except Exception as e: @@ -3737,6 +3831,8 @@ def _register_routes(app: FastAPI): operation_id=result["operation_id"], deduplicated=result.get("deduplicated", False), ) + except OperationValidationError as e: + raise HTTPException(status_code=e.status_code, detail=e.reason) except (AuthenticationError, HTTPException): raise except Exception as e: @@ -4024,6 +4120,8 @@ def _register_routes(app: FastAPI): await app.state.memory.delete_bank(bank_id, fact_type=type, request_context=request_context) return DeleteResponse(success=True) + except OperationValidationError as e: + raise HTTPException(status_code=e.status_code, detail=e.reason) except (AuthenticationError, HTTPException): raise except Exception as e: diff --git a/hindsight-api/hindsight_api/api/mcp.py b/hindsight-api/hindsight_api/api/mcp.py index 9013adae..8a029f22 100644 --- a/hindsight-api/hindsight_api/api/mcp.py +++ b/hindsight-api/hindsight_api/api/mcp.py @@ -166,7 +166,6 @@ def create_mcp_server(memory: MemoryEngine, multi_bank: bool = True) -> FastMCP: api_key_id_resolver=get_current_api_key_id, # Propagate api_key_id for usage metering include_bank_id_param=multi_bank, tools=base_tools, - retain_fire_and_forget=False, # HTTP MCP supports sync/async modes ) register_mcp_tools(mcp, memory, config) diff --git a/hindsight-api/hindsight_api/engine/memory_engine.py b/hindsight-api/hindsight_api/engine/memory_engine.py index 7ad22eb1..aaed0587 100644 --- a/hindsight-api/hindsight_api/engine/memory_engine.py +++ b/hindsight-api/hindsight_api/engine/memory_engine.py @@ -3120,6 +3120,11 @@ class MemoryEngine(MemoryEngineInterface): Dictionary with document info or None if not found """ await self._authenticate_tenant(request_context) + if self._operation_validator: + from hindsight_api.extensions import BankReadContext + + ctx = BankReadContext(bank_id=bank_id, operation="get_document", request_context=request_context) + await self._validate_operation(self._operation_validator.validate_bank_read(ctx)) pool = await self._get_pool() async with acquire_with_retry(pool) as conn: doc = await conn.fetchrow( @@ -3168,6 +3173,11 @@ class MemoryEngine(MemoryEngineInterface): Dictionary with counts of deleted items """ await self._authenticate_tenant(request_context) + if self._operation_validator: + from hindsight_api.extensions import BankWriteContext + + ctx = BankWriteContext(bank_id=bank_id, operation="delete_document", request_context=request_context) + await self._validate_operation(self._operation_validator.validate_bank_write(ctx)) pool = await self._get_pool() invalidated_obs = 0 async with acquire_with_retry(pool) as conn: @@ -3292,6 +3302,11 @@ class MemoryEngine(MemoryEngineInterface): Dictionary with counts of deleted items """ await self._authenticate_tenant(request_context) + if self._operation_validator: + from hindsight_api.extensions import BankWriteContext + + ctx = BankWriteContext(bank_id=bank_id, operation="delete_bank", request_context=request_context) + await self._validate_operation(self._operation_validator.validate_bank_write(ctx)) pool = await self._get_pool() invalidated_obs = 0 result: dict[str, int] = {} @@ -3385,6 +3400,11 @@ class MemoryEngine(MemoryEngineInterface): Dictionary with count of deleted observations """ await self._authenticate_tenant(request_context) + if self._operation_validator: + from hindsight_api.extensions import BankWriteContext + + ctx = BankWriteContext(bank_id=bank_id, operation="clear_observations", request_context=request_context) + await self._validate_operation(self._operation_validator.validate_bank_write(ctx)) pool = await self._get_pool() async with acquire_with_retry(pool) as conn: async with conn.transaction(): @@ -3438,6 +3458,13 @@ class MemoryEngine(MemoryEngineInterface): Dictionary with count of deleted observations """ await self._authenticate_tenant(request_context) + if self._operation_validator: + from hindsight_api.extensions import BankWriteContext + + ctx = BankWriteContext( + bank_id=bank_id, operation="clear_observations_for_memory", request_context=request_context + ) + await self._validate_operation(self._operation_validator.validate_bank_write(ctx)) pool = await self._get_pool() deleted_count = 0 @@ -3484,6 +3511,11 @@ class MemoryEngine(MemoryEngineInterface): Dictionary with consolidation stats """ await self._authenticate_tenant(request_context) + if self._operation_validator: + from hindsight_api.extensions import BankWriteContext + + ctx = BankWriteContext(bank_id=bank_id, operation="run_consolidation", request_context=request_context) + await self._validate_operation(self._operation_validator.validate_bank_write(ctx)) from .consolidation import run_consolidation_job @@ -3529,6 +3561,11 @@ class MemoryEngine(MemoryEngineInterface): Dict with nodes, edges, table_rows, total_units, and limit """ await self._authenticate_tenant(request_context) + if self._operation_validator: + from hindsight_api.extensions import BankReadContext + + ctx = BankReadContext(bank_id=bank_id, operation="get_graph_data", request_context=request_context) + await self._validate_operation(self._operation_validator.validate_bank_read(ctx)) pool = await self._get_pool() async with acquire_with_retry(pool) as conn: # Get memory units, optionally filtered by bank_id and fact_type @@ -3890,6 +3927,11 @@ class MemoryEngine(MemoryEngineInterface): Dict with items (list of memory units) and total count """ await self._authenticate_tenant(request_context) + if self._operation_validator: + from hindsight_api.extensions import BankReadContext + + ctx = BankReadContext(bank_id=bank_id, operation="list_memory_units", request_context=request_context) + await self._validate_operation(self._operation_validator.validate_bank_read(ctx)) pool = await self._get_pool() async with acquire_with_retry(pool) as conn: # Build query conditions @@ -4012,6 +4054,11 @@ class MemoryEngine(MemoryEngineInterface): Dict with memory unit data or None if not found """ await self._authenticate_tenant(request_context) + if self._operation_validator: + from hindsight_api.extensions import BankReadContext + + ctx = BankReadContext(bank_id=bank_id, operation="get_memory_unit", request_context=request_context) + await self._validate_operation(self._operation_validator.validate_bank_read(ctx)) pool = await self._get_pool() async with acquire_with_retry(pool) as conn: # Get the memory unit (include source_memory_ids for mental models) @@ -4123,6 +4170,11 @@ class MemoryEngine(MemoryEngineInterface): Dict with items (list of documents without original_text) and total count """ await self._authenticate_tenant(request_context) + if self._operation_validator: + from hindsight_api.extensions import BankReadContext + + ctx = BankReadContext(bank_id=bank_id, operation="list_documents", request_context=request_context) + await self._validate_operation(self._operation_validator.validate_bank_read(ctx)) pool = await self._get_pool() async with acquire_with_retry(pool) as conn: # Build query conditions @@ -4269,6 +4321,12 @@ class MemoryEngine(MemoryEngineInterface): if not chunk: return None + if self._operation_validator: + from hindsight_api.extensions import BankReadContext + + ctx = BankReadContext(bank_id=chunk["bank_id"], operation="get_chunk", request_context=request_context) + await self._validate_operation(self._operation_validator.validate_bank_read(ctx)) + return { "chunk_id": chunk["chunk_id"], "document_id": chunk["document_id"], @@ -4298,6 +4356,11 @@ class MemoryEngine(MemoryEngineInterface): Dict with name, disposition traits, and mission """ await self._authenticate_tenant(request_context) + if self._operation_validator: + from hindsight_api.extensions import BankReadContext + + ctx = BankReadContext(bank_id=bank_id, operation="get_bank_profile", request_context=request_context) + await self._validate_operation(self._operation_validator.validate_bank_read(ctx)) pool = await self._get_pool() profile = await bank_utils.get_bank_profile(pool, bank_id) @@ -4340,6 +4403,13 @@ class MemoryEngine(MemoryEngineInterface): request_context: Request context for authentication. """ await self._authenticate_tenant(request_context) + if self._operation_validator: + from hindsight_api.extensions import BankWriteContext + + ctx = BankWriteContext( + bank_id=bank_id, operation="update_bank_disposition", request_context=request_context + ) + await self._validate_operation(self._operation_validator.validate_bank_write(ctx)) pool = await self._get_pool() await bank_utils.update_bank_disposition(pool, bank_id, disposition) @@ -4362,6 +4432,11 @@ class MemoryEngine(MemoryEngineInterface): Dict with bank_id and mission. """ await self._authenticate_tenant(request_context) + if self._operation_validator: + from hindsight_api.extensions import BankWriteContext + + ctx = BankWriteContext(bank_id=bank_id, operation="set_bank_mission", request_context=request_context) + await self._validate_operation(self._operation_validator.validate_bank_write(ctx)) pool = await self._get_pool() await bank_utils.set_bank_mission(pool, bank_id, mission) return {"bank_id": bank_id, "mission": mission} @@ -4386,6 +4461,11 @@ class MemoryEngine(MemoryEngineInterface): Dict with 'mission' (str) key """ await self._authenticate_tenant(request_context) + if self._operation_validator: + from hindsight_api.extensions import BankWriteContext + + ctx = BankWriteContext(bank_id=bank_id, operation="merge_bank_mission", request_context=request_context) + await self._validate_operation(self._operation_validator.validate_bank_write(ctx)) pool = await self._get_pool() return await bank_utils.merge_bank_mission(pool, self._reflect_llm_config, bank_id, new_info) @@ -4405,7 +4485,15 @@ class MemoryEngine(MemoryEngineInterface): """ await self._authenticate_tenant(request_context) pool = await self._get_pool() - return await bank_utils.list_banks(pool) + banks = await bank_utils.list_banks(pool) + if self._operation_validator: + from hindsight_api.extensions import BankListContext + + result = await self._operation_validator.filter_bank_list( + BankListContext(banks=banks, request_context=request_context) + ) + banks = result.banks + return banks # ==================== Reflect Methods ==================== @@ -4813,6 +4901,11 @@ class MemoryEngine(MemoryEngineInterface): Dict with items, total, limit, offset """ await self._authenticate_tenant(request_context) + if self._operation_validator: + from hindsight_api.extensions import BankReadContext + + ctx = BankReadContext(bank_id=bank_id, operation="list_entities", request_context=request_context) + await self._validate_operation(self._operation_validator.validate_bank_read(ctx)) pool = await self._get_pool() async with acquire_with_retry(pool) as conn: # Get total count @@ -4900,6 +4993,11 @@ class MemoryEngine(MemoryEngineInterface): Dict with items (list of {tag, count}), total, limit, offset """ await self._authenticate_tenant(request_context) + if self._operation_validator: + from hindsight_api.extensions import BankReadContext + + ctx = BankReadContext(bank_id=bank_id, operation="list_tags", request_context=request_context) + await self._validate_operation(self._operation_validator.validate_bank_read(ctx)) pool = await self._get_pool() async with acquire_with_retry(pool) as conn: # Build pattern filter if provided (convert * to % for ILIKE) @@ -4976,6 +5074,11 @@ class MemoryEngine(MemoryEngineInterface): EntityState with empty observations (summaries now in mental models) """ await self._authenticate_tenant(request_context) + if self._operation_validator: + from hindsight_api.extensions import BankReadContext + + ctx = BankReadContext(bank_id=bank_id, operation="get_entity_state", request_context=request_context) + await self._validate_operation(self._operation_validator.validate_bank_read(ctx)) return EntityState(entity_id=entity_id, canonical_name=entity_name, observations=[]) # ========================================================================= @@ -4990,6 +5093,11 @@ class MemoryEngine(MemoryEngineInterface): ) -> dict[str, Any]: """Get statistics about memory nodes and links for a bank.""" await self._authenticate_tenant(request_context) + if self._operation_validator: + from hindsight_api.extensions import BankReadContext + + ctx = BankReadContext(bank_id=bank_id, operation="get_bank_stats", request_context=request_context) + await self._validate_operation(self._operation_validator.validate_bank_read(ctx)) pool = await self._get_pool() async with acquire_with_retry(pool) as conn: @@ -5072,6 +5180,11 @@ class MemoryEngine(MemoryEngineInterface): ) -> dict[str, Any] | None: """Get entity details including metadata and observations.""" await self._authenticate_tenant(request_context) + if self._operation_validator: + from hindsight_api.extensions import BankReadContext + + ctx = BankReadContext(bank_id=bank_id, operation="get_entity", request_context=request_context) + await self._validate_operation(self._operation_validator.validate_bank_read(ctx)) pool = await self._get_pool() async with acquire_with_retry(pool) as conn: @@ -5440,6 +5553,11 @@ class MemoryEngine(MemoryEngineInterface): List of pinned mental model dicts """ await self._authenticate_tenant(request_context) + if self._operation_validator: + from hindsight_api.extensions import BankReadContext + + ctx = BankReadContext(bank_id=bank_id, operation="list_mental_models", request_context=request_context) + await self._validate_operation(self._operation_validator.validate_bank_read(ctx)) pool = await self._get_pool() async with acquire_with_retry(pool) as conn: @@ -5568,6 +5686,11 @@ class MemoryEngine(MemoryEngineInterface): The created pinned mental model dict """ await self._authenticate_tenant(request_context) + if self._operation_validator: + from hindsight_api.extensions import BankWriteContext + + ctx = BankWriteContext(bank_id=bank_id, operation="create_mental_model", request_context=request_context) + await self._validate_operation(self._operation_validator.validate_bank_write(ctx)) pool = await self._get_pool() # Generate embedding for the content @@ -5745,6 +5868,11 @@ class MemoryEngine(MemoryEngineInterface): Updated pinned mental model dict or None if not found """ await self._authenticate_tenant(request_context) + if self._operation_validator: + from hindsight_api.extensions import BankWriteContext + + ctx = BankWriteContext(bank_id=bank_id, operation="update_mental_model", request_context=request_context) + await self._validate_operation(self._operation_validator.validate_bank_write(ctx)) pool = await self._get_pool() async with acquire_with_retry(pool) as conn: @@ -5830,6 +5958,11 @@ class MemoryEngine(MemoryEngineInterface): True if deleted, False if not found """ await self._authenticate_tenant(request_context) + if self._operation_validator: + from hindsight_api.extensions import BankWriteContext + + ctx = BankWriteContext(bank_id=bank_id, operation="delete_mental_model", request_context=request_context) + await self._validate_operation(self._operation_validator.validate_bank_write(ctx)) pool = await self._get_pool() async with acquire_with_retry(pool) as conn: @@ -5904,6 +6037,11 @@ class MemoryEngine(MemoryEngineInterface): List of directive dicts """ await self._authenticate_tenant(request_context) + if self._operation_validator: + from hindsight_api.extensions import BankReadContext + + ctx = BankReadContext(bank_id=bank_id, operation="list_directives", request_context=request_context) + await self._validate_operation(self._operation_validator.validate_bank_read(ctx)) pool = await self._get_pool() async with acquire_with_retry(pool) as conn: @@ -5970,6 +6108,11 @@ class MemoryEngine(MemoryEngineInterface): Directive dict or None if not found """ await self._authenticate_tenant(request_context) + if self._operation_validator: + from hindsight_api.extensions import BankReadContext + + ctx = BankReadContext(bank_id=bank_id, operation="get_directive", request_context=request_context) + await self._validate_operation(self._operation_validator.validate_bank_read(ctx)) pool = await self._get_pool() async with acquire_with_retry(pool) as conn: @@ -6011,6 +6154,11 @@ class MemoryEngine(MemoryEngineInterface): The created directive dict """ await self._authenticate_tenant(request_context) + if self._operation_validator: + from hindsight_api.extensions import BankWriteContext + + ctx = BankWriteContext(bank_id=bank_id, operation="create_directive", request_context=request_context) + await self._validate_operation(self._operation_validator.validate_bank_write(ctx)) pool = await self._get_pool() async with acquire_with_retry(pool) as conn: @@ -6060,6 +6208,11 @@ class MemoryEngine(MemoryEngineInterface): Updated directive dict or None if not found """ await self._authenticate_tenant(request_context) + if self._operation_validator: + from hindsight_api.extensions import BankWriteContext + + ctx = BankWriteContext(bank_id=bank_id, operation="update_directive", request_context=request_context) + await self._validate_operation(self._operation_validator.validate_bank_write(ctx)) pool = await self._get_pool() # Build update query dynamically @@ -6125,6 +6278,11 @@ class MemoryEngine(MemoryEngineInterface): True if deleted, False if not found """ await self._authenticate_tenant(request_context) + if self._operation_validator: + from hindsight_api.extensions import BankWriteContext + + ctx = BankWriteContext(bank_id=bank_id, operation="delete_directive", request_context=request_context) + await self._validate_operation(self._operation_validator.validate_bank_write(ctx)) pool = await self._get_pool() async with acquire_with_retry(pool) as conn: @@ -6172,6 +6330,11 @@ class MemoryEngine(MemoryEngineInterface): Dict with total count and list of operations, sorted by most recent first """ await self._authenticate_tenant(request_context) + if self._operation_validator: + from hindsight_api.extensions import BankReadContext + + ctx = BankReadContext(bank_id=bank_id, operation="list_operations", request_context=request_context) + await self._validate_operation(self._operation_validator.validate_bank_read(ctx)) pool = await self._get_pool() async with acquire_with_retry(pool) as conn: @@ -6253,6 +6416,11 @@ class MemoryEngine(MemoryEngineInterface): - child_operations: (for parent operations) list of child operation statuses """ await self._authenticate_tenant(request_context) + if self._operation_validator: + from hindsight_api.extensions import BankReadContext + + ctx = BankReadContext(bank_id=bank_id, operation="get_operation_status", request_context=request_context) + await self._validate_operation(self._operation_validator.validate_bank_read(ctx)) pool = await self._get_pool() op_uuid = uuid.UUID(operation_id) @@ -6382,6 +6550,11 @@ class MemoryEngine(MemoryEngineInterface): ) -> dict[str, Any]: """Cancel a pending async operation.""" await self._authenticate_tenant(request_context) + if self._operation_validator: + from hindsight_api.extensions import BankWriteContext + + ctx = BankWriteContext(bank_id=bank_id, operation="cancel_operation", request_context=request_context) + await self._validate_operation(self._operation_validator.validate_bank_write(ctx)) pool = await self._get_pool() op_uuid = uuid.UUID(operation_id) @@ -6417,6 +6590,11 @@ class MemoryEngine(MemoryEngineInterface): ) -> dict[str, Any]: """Update bank name and/or mission.""" await self._authenticate_tenant(request_context) + if self._operation_validator: + from hindsight_api.extensions import BankWriteContext + + ctx = BankWriteContext(bank_id=bank_id, operation="update_bank", request_context=request_context) + await self._validate_operation(self._operation_validator.validate_bank_write(ctx)) pool = await self._get_pool() async with acquire_with_retry(pool) as conn: @@ -6544,6 +6722,17 @@ class MemoryEngine(MemoryEngineInterface): """ await self._authenticate_tenant(request_context) + # Run operation validator (bank access, credits, etc.) before queuing + if self._operation_validator: + from hindsight_api.extensions import RetainContext + + ctx = RetainContext( + bank_id=bank_id, + contents=[dict(c) for c in contents], + request_context=request_context, + ) + await self._validate_operation(self._operation_validator.validate_retain(ctx)) + # Validate no duplicate document_ids in the batch # Having duplicate document_ids causes race conditions in document upserts during parallel processing doc_ids = [item.get("document_id") for item in contents if item.get("document_id")] @@ -6787,6 +6976,13 @@ class MemoryEngine(MemoryEngineInterface): Dict with operation_id """ await self._authenticate_tenant(request_context) + if self._operation_validator: + from hindsight_api.extensions import BankWriteContext + + ctx = BankWriteContext( + bank_id=bank_id, operation="submit_async_consolidation", request_context=request_context + ) + await self._validate_operation(self._operation_validator.validate_bank_write(ctx)) # Pass tenant_id and api_key_id through task payload so the worker # can provide request context to extension hooks (e.g., usage metering diff --git a/hindsight-api/hindsight_api/extensions/__init__.py b/hindsight-api/hindsight_api/extensions/__init__.py index b60f515f..0e93225a 100644 --- a/hindsight-api/hindsight_api/extensions/__init__.py +++ b/hindsight-api/hindsight_api/extensions/__init__.py @@ -22,6 +22,11 @@ from hindsight_api.extensions.http import HttpExtension from hindsight_api.extensions.loader import load_extension from hindsight_api.extensions.mcp import MCPExtension from hindsight_api.extensions.operation_validator import ( + # Bank Management operations + BankListContext, + BankListResult, + BankReadContext, + BankWriteContext, # Consolidation operation ConsolidateContext, ConsolidateResult, @@ -70,6 +75,11 @@ __all__ = [ "RetainContext", "RetainResult", "ValidationResult", + # Operation Validator - Bank Management + "BankListContext", + "BankListResult", + "BankReadContext", + "BankWriteContext", # Operation Validator - Consolidation "ConsolidateContext", "ConsolidateResult", diff --git a/hindsight-api/hindsight_api/extensions/operation_validator.py b/hindsight-api/hindsight_api/extensions/operation_validator.py index 15eb1b22..56334418 100644 --- a/hindsight-api/hindsight_api/extensions/operation_validator.py +++ b/hindsight-api/hindsight_api/extensions/operation_validator.py @@ -200,6 +200,44 @@ class ConsolidateResult: error: str | None = None +# ============================================================================= +# Bank Management Contexts +# ============================================================================= + + +@dataclass +class BankReadContext: + """Context for a bank read operation validation (pre-operation).""" + + bank_id: str + operation: str # "get_bank_profile", "get_bank_stats" + request_context: "RequestContext" + + +@dataclass +class BankWriteContext: + """Context for a bank write operation validation (pre-operation).""" + + bank_id: str + operation: str # "delete_bank", "update_bank", "update_bank_disposition", "set_bank_mission", "merge_bank_mission", "clear_observations", "clear_observations_for_memory" + request_context: "RequestContext" + + +@dataclass +class BankListContext: + """Context for filtering the bank list (post-query).""" + + banks: list[dict] + request_context: "RequestContext" + + +@dataclass +class BankListResult: + """Result of filtering the bank list.""" + + banks: list[dict] + + # ============================================================================= # Mental Model Contexts # ============================================================================= @@ -535,3 +573,63 @@ class OperationValidatorExtension(Extension, ABC): - error: Error message (if failed) """ pass + + # ========================================================================= + # Bank Management - Validation hooks (optional - override to implement) + # ========================================================================= + + async def validate_bank_read(self, ctx: BankReadContext) -> ValidationResult: + """ + Validate a bank read operation before execution. + + Override to implement custom validation logic for bank reads + (get_bank_profile, get_bank_stats). + + Args: + ctx: Context containing: + - bank_id: Bank identifier + - operation: Operation name + - request_context: Request context with auth info + + Returns: + ValidationResult indicating whether the operation is allowed. + """ + return ValidationResult.accept() + + async def validate_bank_write(self, ctx: BankWriteContext) -> ValidationResult: + """ + Validate a bank write operation before execution. + + Override to implement custom validation logic for bank writes + (delete_bank, update_bank, update_bank_disposition, set_bank_mission, + merge_bank_mission, clear_observations, clear_observations_for_memory). + + Args: + ctx: Context containing: + - bank_id: Bank identifier + - operation: Operation name + - request_context: Request context with auth info + + Returns: + ValidationResult indicating whether the operation is allowed. + """ + return ValidationResult.accept() + + async def filter_bank_list(self, ctx: BankListContext) -> BankListResult: + """ + Filter the bank list after querying. + + Unlike validate_* methods, this is a post-query filter that narrows results + rather than a gate that blocks the operation. + + Override to implement custom filtering (e.g., restrict to allowed banks). + + Args: + ctx: Context containing: + - banks: List of bank dicts from the database + - request_context: Request context with auth info + + Returns: + BankListResult with the filtered list of banks. + """ + return BankListResult(banks=ctx.banks) diff --git a/hindsight-api/hindsight_api/mcp_tools.py b/hindsight-api/hindsight_api/mcp_tools.py index cc46434b..bb0b46ae 100644 --- a/hindsight-api/hindsight_api/mcp_tools.py +++ b/hindsight-api/hindsight_api/mcp_tools.py @@ -20,6 +20,7 @@ from hindsight_api.config import ( ) from hindsight_api.engine.memory_engine import Budget from hindsight_api.engine.response_models import VALID_RECALL_FACT_TYPES +from hindsight_api.extensions import OperationValidationError from hindsight_api.models import RequestContext logger = logging.getLogger(__name__) @@ -52,7 +53,6 @@ class MCPToolsConfig: recall_description: str | None = None # Retain behavior - retain_fire_and_forget: bool = False # If True, use asyncio.create_task pattern def _get_request_context(config: MCPToolsConfig) -> RequestContext: @@ -320,106 +320,56 @@ def _register_retain(mcp: FastMCP, memory: MemoryEngine, config: MCPToolsConfig) description = config.retain_description or DEFAULT_MCP_RETAIN_DESCRIPTION if config.include_bank_id_param: - if config.retain_fire_and_forget: - @mcp.tool(description=description) - async def retain( - content: str, - context: str = "general", - timestamp: str | None = None, - tags: list[str] | None = None, - metadata: dict[str, str] | None = None, - document_id: str | None = None, - bank_id: str | None = None, - ) -> dict: - """ - Args: - content: The fact/memory to store (be specific and include relevant details) - context: Category for the memory (e.g., 'preferences', 'work', 'hobbies', 'family'). Default: 'general' - timestamp: When this event/fact occurred (ISO format, e.g., '2024-01-15T10:30:00Z'). Useful for timeline tracking. - tags: Optional tags for scoped visibility filtering (e.g., ['project:alpha', 'user:123']) - metadata: Optional key-value metadata to attach (e.g., {'source': 'slack', 'channel': 'general'}) - document_id: Optional document ID to associate this memory with - bank_id: Optional bank to store in (defaults to session bank). Use for cross-bank operations. - """ - import asyncio + @mcp.tool(description=description) + async def retain( + content: str, + context: str = "general", + timestamp: str | None = None, + tags: list[str] | None = None, + metadata: dict[str, str] | None = None, + document_id: str | None = None, + bank_id: str | None = None, + ) -> dict: + """ + Args: + content: The fact/memory to store (be specific and include relevant details) + context: Category for the memory (e.g., 'preferences', 'work', 'hobbies', 'family'). Default: 'general' + timestamp: When this event/fact occurred (ISO format, e.g., '2024-01-15T10:30:00Z'). Useful for timeline tracking. + tags: Optional tags for scoped visibility filtering (e.g., ['project:alpha', 'user:123']) + metadata: Optional key-value metadata to attach (e.g., {'source': 'slack', 'channel': 'general'}) + document_id: Optional document ID to associate this memory with + bank_id: Optional bank to store in (defaults to session bank). Use for cross-bank operations. + """ + target_bank = bank_id or config.bank_id_resolver() + if target_bank is None: + return {"status": "error", "message": "No bank_id configured"} - target_bank = bank_id or config.bank_id_resolver() - if target_bank is None: - return {"status": "error", "message": "No bank_id configured"} + content_dict, error = build_content_dict(content, context, timestamp, tags, metadata, document_id) + if error: + return {"status": "error", "message": error} - content_dict, error = build_content_dict(content, context, timestamp, tags, metadata, document_id) - if error: - return {"status": "error", "message": error} + request_context = _get_request_context(config) - request_context = _get_request_context(config) - - async def _retain(): - try: - await memory.retain_batch_async( - bank_id=target_bank, - contents=[content_dict], - request_context=request_context, - ) - except Exception as e: - logger.error(f"Error storing memory: {e}", exc_info=True) - - asyncio.create_task(_retain()) - return {"status": "accepted", "message": "Memory storage initiated"} - - else: - - @mcp.tool(description=description) - async def retain( - content: str, - context: str = "general", - timestamp: str | None = None, - tags: list[str] | None = None, - metadata: dict[str, str] | None = None, - document_id: str | None = None, - async_processing: bool = True, - bank_id: str | None = None, - ) -> str: - """ - Args: - content: The fact/memory to store (be specific and include relevant details) - context: Category for the memory (e.g., 'preferences', 'work', 'hobbies', 'family'). Default: 'general' - timestamp: When this event/fact occurred (ISO format, e.g., '2024-01-15T10:30:00Z'). Useful for timeline tracking. - tags: Optional tags for scoped visibility filtering (e.g., ['project:alpha', 'user:123']) - metadata: Optional key-value metadata to attach (e.g., {'source': 'slack', 'channel': 'general'}) - document_id: Optional document ID to associate this memory with - async_processing: If True, queue for background processing and return immediately. If False, wait for completion. Default: True - bank_id: Optional bank to store in (defaults to session bank). Use for cross-bank operations. - """ - try: - target_bank = bank_id or config.bank_id_resolver() - if target_bank is None: - return "Error: No bank_id configured" - - content_dict, error = build_content_dict(content, context, timestamp, tags, metadata, document_id) - if error: - return f"Error: {error}" - - contents = [content_dict] - request_context = _get_request_context(config) - if async_processing: - result = await memory.submit_async_retain( - bank_id=target_bank, contents=contents, request_context=request_context - ) - return f"Memory queued for background processing (operation_id: {result.get('operation_id', 'N/A')})" - else: - await memory.retain_batch_async( - bank_id=target_bank, - contents=contents, - request_context=request_context, - ) - return f"Memory stored successfully in bank '{target_bank}'" - except Exception as e: - logger.error(f"Error storing memory: {e}", exc_info=True) - return f"Error: {str(e)}" + try: + result = await memory.submit_async_retain( + bank_id=target_bank, + contents=[content_dict], + request_context=request_context, + ) + return { + "status": "accepted", + "message": "Memory storage initiated", + "operation_id": result.get("operation_id"), + } + except OperationValidationError as e: + logger.warning(f"Retain rejected: {e}") + return {"status": "error", "message": str(e)} + except Exception as e: + logger.error(f"Error storing memory: {e}", exc_info=True) + return {"status": "error", "message": str(e)} else: - # No bank_id param - use fixed bank from resolver @mcp.tool(description=description) async def retain( @@ -439,8 +389,6 @@ def _register_retain(mcp: FastMCP, memory: MemoryEngine, config: MCPToolsConfig) metadata: Optional key-value metadata to attach (e.g., {'source': 'slack', 'channel': 'general'}) document_id: Optional document ID to associate this memory with """ - import asyncio - target_bank = config.bank_id_resolver() if target_bank is None: return {"status": "error", "message": "No bank_id configured"} @@ -451,18 +399,23 @@ def _register_retain(mcp: FastMCP, memory: MemoryEngine, config: MCPToolsConfig) request_context = _get_request_context(config) - async def _retain(): - try: - await memory.retain_batch_async( - bank_id=target_bank, - contents=[content_dict], - request_context=request_context, - ) - except Exception as e: - logger.error(f"Error storing memory: {e}", exc_info=True) - - asyncio.create_task(_retain()) - return {"status": "accepted", "message": "Memory storage initiated"} + try: + result = await memory.submit_async_retain( + bank_id=target_bank, + contents=[content_dict], + request_context=request_context, + ) + return { + "status": "accepted", + "message": "Memory storage initiated", + "operation_id": result.get("operation_id"), + } + except OperationValidationError as e: + logger.warning(f"Retain rejected: {e}") + return {"status": "error", "message": str(e)} + except Exception as e: + logger.error(f"Error storing memory: {e}", exc_info=True) + return {"status": "error", "message": str(e)} def _register_recall(mcp: FastMCP, memory: MemoryEngine, config: MCPToolsConfig) -> None: @@ -519,6 +472,9 @@ def _register_recall(mcp: FastMCP, memory: MemoryEngine, config: MCPToolsConfig) recall_result = await memory.recall_async(**recall_kwargs) return recall_result.model_dump_json(indent=2) + except OperationValidationError as e: + logger.warning(f"Recall rejected: {e}") + return json.dumps({"error": str(e), "results": []}) except ValueError as e: return f'{{"error": "{e}", "results": []}}' except Exception as e: @@ -573,6 +529,9 @@ def _register_recall(mcp: FastMCP, memory: MemoryEngine, config: MCPToolsConfig) recall_result = await memory.recall_async(**recall_kwargs) return recall_result.model_dump() + except OperationValidationError as e: + logger.warning(f"Recall rejected: {e}") + return {"error": str(e), "results": []} except ValueError as e: return {"error": str(e), "results": []} except Exception as e: @@ -653,6 +612,9 @@ def _register_reflect(mcp: FastMCP, memory: MemoryEngine, config: MCPToolsConfig if response_schema is not None and hasattr(reflect_result, "structured_output"): result_data["structured_output"] = reflect_result.structured_output return json.dumps(result_data, indent=2) + except OperationValidationError as e: + logger.warning(f"Reflect rejected: {e}") + return json.dumps({"error": str(e)}) except Exception as e: logger.error(f"Error reflecting: {e}", exc_info=True) return f'{{"error": "{e}", "text": ""}}' @@ -725,6 +687,9 @@ def _register_reflect(mcp: FastMCP, memory: MemoryEngine, config: MCPToolsConfig if response_schema is not None and hasattr(reflect_result, "structured_output"): result_data["structured_output"] = reflect_result.structured_output return result_data + except OperationValidationError as e: + logger.warning(f"Reflect rejected: {e}") + return {"error": str(e)} except Exception as e: logger.error(f"Error reflecting: {e}", exc_info=True) return {"error": str(e), "text": ""} @@ -747,6 +712,9 @@ def _register_list_banks(mcp: FastMCP, memory: MemoryEngine, config: MCPToolsCon try: banks = await memory.list_banks(request_context=_get_request_context(config)) return json.dumps({"banks": banks}, indent=2) + except OperationValidationError as e: + logger.warning(f"Operation rejected: {e}") + return json.dumps({"error": str(e), "banks": []}) except Exception as e: logger.error(f"Error listing banks: {e}", exc_info=True) return f'{{"error": "{e}", "banks": []}}' @@ -788,6 +756,9 @@ def _register_create_bank(mcp: FastMCP, memory: MemoryEngine, config: MCPToolsCo if "disposition" in profile and hasattr(profile["disposition"], "model_dump"): profile["disposition"] = profile["disposition"].model_dump() return json.dumps(profile, indent=2) + except OperationValidationError as e: + logger.warning(f"Operation rejected: {e}") + return json.dumps({"error": str(e)}) except Exception as e: logger.error(f"Error creating bank: {e}", exc_info=True) return f'{{"error": "{e}"}}' @@ -843,6 +814,9 @@ def _register_list_mental_models(mcp: FastMCP, memory: MemoryEngine, config: MCP request_context=_get_request_context(config), ) return json.dumps({"items": models}, indent=2, default=str) + except OperationValidationError as e: + logger.warning(f"Operation rejected: {e}") + return json.dumps({"error": str(e)}) except Exception as e: logger.error(f"Error listing mental models: {e}", exc_info=True) return f'{{"error": "{e}", "items": []}}' @@ -874,6 +848,9 @@ def _register_list_mental_models(mcp: FastMCP, memory: MemoryEngine, config: MCP request_context=_get_request_context(config), ) return {"items": models} + except OperationValidationError as e: + logger.warning(f"Operation rejected: {e}") + return {"error": str(e)} except Exception as e: logger.error(f"Error listing mental models: {e}", exc_info=True) return {"error": str(e), "items": []} @@ -912,6 +889,9 @@ def _register_get_mental_model(mcp: FastMCP, memory: MemoryEngine, config: MCPTo if model is None: return json.dumps({"error": f"Mental model '{mental_model_id}' not found in bank '{target_bank}'"}) return json.dumps(model, indent=2, default=str) + except OperationValidationError as e: + logger.warning(f"Operation rejected: {e}") + return json.dumps({"error": str(e)}) except Exception as e: logger.error(f"Error getting mental model: {e}", exc_info=True) return f'{{"error": "{e}"}}' @@ -944,6 +924,9 @@ def _register_get_mental_model(mcp: FastMCP, memory: MemoryEngine, config: MCPTo if model is None: return {"error": f"Mental model '{mental_model_id}' not found in bank '{target_bank}'"} return model + except OperationValidationError as e: + logger.warning(f"Operation rejected: {e}") + return {"error": str(e)} except Exception as e: logger.error(f"Error getting mental model: {e}", exc_info=True) return {"error": str(e)} @@ -1027,6 +1010,9 @@ def _register_create_mental_model(mcp: FastMCP, memory: MemoryEngine, config: MC "message": f"Mental model '{name}' created. Content is being generated asynchronously.", } ) + except OperationValidationError as e: + logger.warning(f"Operation rejected: {e}") + return json.dumps({"error": str(e)}) except ValueError as e: return json.dumps({"error": str(e)}) except Exception as e: @@ -1102,6 +1088,9 @@ def _register_create_mental_model(mcp: FastMCP, memory: MemoryEngine, config: MC "status": "created", "message": f"Mental model '{name}' created. Content is being generated asynchronously.", } + except OperationValidationError as e: + logger.warning(f"Operation rejected: {e}") + return {"error": str(e)} except ValueError as e: return {"error": str(e)} except Exception as e: @@ -1166,6 +1155,9 @@ def _register_update_mental_model(mcp: FastMCP, memory: MemoryEngine, config: MC if model is None: return json.dumps({"error": f"Mental model '{mental_model_id}' not found in bank '{target_bank}'"}) return json.dumps(model, indent=2, default=str) + except OperationValidationError as e: + logger.warning(f"Operation rejected: {e}") + return json.dumps({"error": str(e)}) except Exception as e: logger.error(f"Error updating mental model: {e}", exc_info=True) return f'{{"error": "{e}"}}' @@ -1222,6 +1214,9 @@ def _register_update_mental_model(mcp: FastMCP, memory: MemoryEngine, config: MC if model is None: return {"error": f"Mental model '{mental_model_id}' not found in bank '{target_bank}'"} return model + except OperationValidationError as e: + logger.warning(f"Operation rejected: {e}") + return {"error": str(e)} except Exception as e: logger.error(f"Error updating mental model: {e}", exc_info=True) return {"error": str(e)} @@ -1259,6 +1254,9 @@ def _register_delete_mental_model(mcp: FastMCP, memory: MemoryEngine, config: MC if not deleted: return json.dumps({"error": f"Mental model '{mental_model_id}' not found in bank '{target_bank}'"}) return json.dumps({"status": "deleted", "mental_model_id": mental_model_id}) + except OperationValidationError as e: + logger.warning(f"Operation rejected: {e}") + return json.dumps({"error": str(e)}) except Exception as e: logger.error(f"Error deleting mental model: {e}", exc_info=True) return f'{{"error": "{e}"}}' @@ -1290,6 +1288,9 @@ def _register_delete_mental_model(mcp: FastMCP, memory: MemoryEngine, config: MC if not deleted: return {"error": f"Mental model '{mental_model_id}' not found in bank '{target_bank}'"} return {"status": "deleted", "mental_model_id": mental_model_id} + except OperationValidationError as e: + logger.warning(f"Operation rejected: {e}") + return {"error": str(e)} except Exception as e: logger.error(f"Error deleting mental model: {e}", exc_info=True) return {"error": str(e)} @@ -1333,6 +1334,9 @@ def _register_refresh_mental_model(mcp: FastMCP, memory: MemoryEngine, config: M "message": f"Refresh queued for mental model '{mental_model_id}'.", } ) + except OperationValidationError as e: + logger.warning(f"Operation rejected: {e}") + return json.dumps({"error": str(e)}) except ValueError as e: return json.dumps({"error": str(e)}) except Exception as e: @@ -1370,6 +1374,9 @@ def _register_refresh_mental_model(mcp: FastMCP, memory: MemoryEngine, config: M "status": "queued", "message": f"Refresh queued for mental model '{mental_model_id}'.", } + except OperationValidationError as e: + logger.warning(f"Operation rejected: {e}") + return {"error": str(e)} except ValueError as e: return {"error": str(e)} except Exception as e: @@ -1416,6 +1423,9 @@ def _register_list_directives(mcp: FastMCP, memory: MemoryEngine, config: MCPToo request_context=_get_request_context(config), ) return json.dumps({"items": directives}, indent=2, default=str) + except OperationValidationError as e: + logger.warning(f"Operation rejected: {e}") + return json.dumps({"error": str(e)}) except Exception as e: logger.error(f"Error listing directives: {e}", exc_info=True) return f'{{"error": "{e}", "items": []}}' @@ -1449,6 +1459,9 @@ def _register_list_directives(mcp: FastMCP, memory: MemoryEngine, config: MCPToo request_context=_get_request_context(config), ) return {"items": directives} + except OperationValidationError as e: + logger.warning(f"Operation rejected: {e}") + return {"error": str(e)} except Exception as e: logger.error(f"Error listing directives: {e}", exc_info=True) return {"error": str(e), "items": []} @@ -1496,6 +1509,9 @@ def _register_create_directive(mcp: FastMCP, memory: MemoryEngine, config: MCPTo request_context=_get_request_context(config), ) return json.dumps(directive, indent=2, default=str) + except OperationValidationError as e: + logger.warning(f"Operation rejected: {e}") + return json.dumps({"error": str(e)}) except Exception as e: logger.error(f"Error creating directive: {e}", exc_info=True) return f'{{"error": "{e}"}}' @@ -1537,6 +1553,9 @@ def _register_create_directive(mcp: FastMCP, memory: MemoryEngine, config: MCPTo request_context=_get_request_context(config), ) return directive + except OperationValidationError as e: + logger.warning(f"Operation rejected: {e}") + return {"error": str(e)} except Exception as e: logger.error(f"Error creating directive: {e}", exc_info=True) return {"error": str(e)} @@ -1574,6 +1593,9 @@ def _register_delete_directive(mcp: FastMCP, memory: MemoryEngine, config: MCPTo if not deleted: return json.dumps({"error": f"Directive '{directive_id}' not found"}) return json.dumps({"status": "deleted", "directive_id": directive_id}) + except OperationValidationError as e: + logger.warning(f"Operation rejected: {e}") + return json.dumps({"error": str(e)}) except Exception as e: logger.error(f"Error deleting directive: {e}", exc_info=True) return f'{{"error": "{e}"}}' @@ -1605,6 +1627,9 @@ def _register_delete_directive(mcp: FastMCP, memory: MemoryEngine, config: MCPTo if not deleted: return {"error": f"Directive '{directive_id}' not found"} return {"status": "deleted", "directive_id": directive_id} + except OperationValidationError as e: + logger.warning(f"Operation rejected: {e}") + return {"error": str(e)} except Exception as e: logger.error(f"Error deleting directive: {e}", exc_info=True) return {"error": str(e)} @@ -1655,6 +1680,9 @@ def _register_list_memories(mcp: FastMCP, memory: MemoryEngine, config: MCPTools request_context=_get_request_context(config), ) return json.dumps(result, indent=2, default=str) + except OperationValidationError as e: + logger.warning(f"Operation rejected: {e}") + return json.dumps({"error": str(e)}) except Exception as e: logger.error(f"Error listing memories: {e}", exc_info=True) return f'{{"error": "{e}"}}' @@ -1694,6 +1722,9 @@ def _register_list_memories(mcp: FastMCP, memory: MemoryEngine, config: MCPTools request_context=_get_request_context(config), ) return result + except OperationValidationError as e: + logger.warning(f"Operation rejected: {e}") + return {"error": str(e)} except Exception as e: logger.error(f"Error listing memories: {e}", exc_info=True) return {"error": str(e)} @@ -1731,6 +1762,9 @@ def _register_get_memory(mcp: FastMCP, memory: MemoryEngine, config: MCPToolsCon if result is None: return json.dumps({"error": f"Memory '{memory_id}' not found"}) return json.dumps(result, indent=2, default=str) + except OperationValidationError as e: + logger.warning(f"Operation rejected: {e}") + return json.dumps({"error": str(e)}) except Exception as e: logger.error(f"Error getting memory: {e}", exc_info=True) return f'{{"error": "{e}"}}' @@ -1762,6 +1796,9 @@ def _register_get_memory(mcp: FastMCP, memory: MemoryEngine, config: MCPToolsCon if result is None: return {"error": f"Memory '{memory_id}' not found"} return result + except OperationValidationError as e: + logger.warning(f"Operation rejected: {e}") + return {"error": str(e)} except Exception as e: logger.error(f"Error getting memory: {e}", exc_info=True) return {"error": str(e)} @@ -1796,6 +1833,9 @@ def _register_delete_memory(mcp: FastMCP, memory: MemoryEngine, config: MCPTools request_context=_get_request_context(config), ) return json.dumps({"status": "deleted", "memory_id": memory_id, **result}, default=str) + except OperationValidationError as e: + logger.warning(f"Operation rejected: {e}") + return json.dumps({"error": str(e)}) except Exception as e: logger.error(f"Error deleting memory: {e}", exc_info=True) return f'{{"error": "{e}"}}' @@ -1824,6 +1864,9 @@ def _register_delete_memory(mcp: FastMCP, memory: MemoryEngine, config: MCPTools request_context=_get_request_context(config), ) return {"status": "deleted", "memory_id": memory_id, **result} + except OperationValidationError as e: + logger.warning(f"Operation rejected: {e}") + return {"error": str(e)} except Exception as e: logger.error(f"Error deleting memory: {e}", exc_info=True) return {"error": str(e)} @@ -1868,6 +1911,9 @@ def _register_list_documents(mcp: FastMCP, memory: MemoryEngine, config: MCPTool request_context=_get_request_context(config), ) return json.dumps(result, indent=2, default=str) + except OperationValidationError as e: + logger.warning(f"Operation rejected: {e}") + return json.dumps({"error": str(e)}) except Exception as e: logger.error(f"Error listing documents: {e}", exc_info=True) return f'{{"error": "{e}"}}' @@ -1901,6 +1947,9 @@ def _register_list_documents(mcp: FastMCP, memory: MemoryEngine, config: MCPTool request_context=_get_request_context(config), ) return result + except OperationValidationError as e: + logger.warning(f"Operation rejected: {e}") + return {"error": str(e)} except Exception as e: logger.error(f"Error listing documents: {e}", exc_info=True) return {"error": str(e)} @@ -1938,6 +1987,9 @@ def _register_get_document(mcp: FastMCP, memory: MemoryEngine, config: MCPToolsC if result is None: return json.dumps({"error": f"Document '{document_id}' not found"}) return json.dumps(result, indent=2, default=str) + except OperationValidationError as e: + logger.warning(f"Operation rejected: {e}") + return json.dumps({"error": str(e)}) except Exception as e: logger.error(f"Error getting document: {e}", exc_info=True) return f'{{"error": "{e}"}}' @@ -1969,6 +2021,9 @@ def _register_get_document(mcp: FastMCP, memory: MemoryEngine, config: MCPToolsC if result is None: return {"error": f"Document '{document_id}' not found"} return result + except OperationValidationError as e: + logger.warning(f"Operation rejected: {e}") + return {"error": str(e)} except Exception as e: logger.error(f"Error getting document: {e}", exc_info=True) return {"error": str(e)} @@ -2004,6 +2059,9 @@ def _register_delete_document(mcp: FastMCP, memory: MemoryEngine, config: MCPToo request_context=_get_request_context(config), ) return json.dumps({"status": "deleted", "document_id": document_id, **result}, default=str) + except OperationValidationError as e: + logger.warning(f"Operation rejected: {e}") + return json.dumps({"error": str(e)}) except Exception as e: logger.error(f"Error deleting document: {e}", exc_info=True) return f'{{"error": "{e}"}}' @@ -2033,6 +2091,9 @@ def _register_delete_document(mcp: FastMCP, memory: MemoryEngine, config: MCPToo request_context=_get_request_context(config), ) return {"status": "deleted", "document_id": document_id, **result} + except OperationValidationError as e: + logger.warning(f"Operation rejected: {e}") + return {"error": str(e)} except Exception as e: logger.error(f"Error deleting document: {e}", exc_info=True) return {"error": str(e)} @@ -2076,6 +2137,9 @@ def _register_list_operations(mcp: FastMCP, memory: MemoryEngine, config: MCPToo request_context=_get_request_context(config), ) return json.dumps(result, indent=2, default=str) + except OperationValidationError as e: + logger.warning(f"Operation rejected: {e}") + return json.dumps({"error": str(e)}) except Exception as e: logger.error(f"Error listing operations: {e}", exc_info=True) return f'{{"error": "{e}"}}' @@ -2108,6 +2172,9 @@ def _register_list_operations(mcp: FastMCP, memory: MemoryEngine, config: MCPToo request_context=_get_request_context(config), ) return result + except OperationValidationError as e: + logger.warning(f"Operation rejected: {e}") + return {"error": str(e)} except Exception as e: logger.error(f"Error listing operations: {e}", exc_info=True) return {"error": str(e)} @@ -2143,6 +2210,9 @@ def _register_get_operation(mcp: FastMCP, memory: MemoryEngine, config: MCPTools request_context=_get_request_context(config), ) return json.dumps(result, indent=2, default=str) + except OperationValidationError as e: + logger.warning(f"Operation rejected: {e}") + return json.dumps({"error": str(e)}) except Exception as e: logger.error(f"Error getting operation: {e}", exc_info=True) return f'{{"error": "{e}"}}' @@ -2172,6 +2242,9 @@ def _register_get_operation(mcp: FastMCP, memory: MemoryEngine, config: MCPTools request_context=_get_request_context(config), ) return result + except OperationValidationError as e: + logger.warning(f"Operation rejected: {e}") + return {"error": str(e)} except Exception as e: logger.error(f"Error getting operation: {e}", exc_info=True) return {"error": str(e)} @@ -2205,6 +2278,9 @@ def _register_cancel_operation(mcp: FastMCP, memory: MemoryEngine, config: MCPTo request_context=_get_request_context(config), ) return json.dumps(result, indent=2, default=str) + except OperationValidationError as e: + logger.warning(f"Operation rejected: {e}") + return json.dumps({"error": str(e)}) except Exception as e: logger.error(f"Error cancelling operation: {e}", exc_info=True) return f'{{"error": "{e}"}}' @@ -2232,6 +2308,9 @@ def _register_cancel_operation(mcp: FastMCP, memory: MemoryEngine, config: MCPTo request_context=_get_request_context(config), ) return result + except OperationValidationError as e: + logger.warning(f"Operation rejected: {e}") + return {"error": str(e)} except Exception as e: logger.error(f"Error cancelling operation: {e}", exc_info=True) return {"error": str(e)} @@ -2275,6 +2354,9 @@ def _register_list_tags(mcp: FastMCP, memory: MemoryEngine, config: MCPToolsConf request_context=_get_request_context(config), ) return json.dumps(result, indent=2, default=str) + except OperationValidationError as e: + logger.warning(f"Operation rejected: {e}") + return json.dumps({"error": str(e)}) except Exception as e: logger.error(f"Error listing tags: {e}", exc_info=True) return f'{{"error": "{e}"}}' @@ -2307,6 +2389,9 @@ def _register_list_tags(mcp: FastMCP, memory: MemoryEngine, config: MCPToolsConf request_context=_get_request_context(config), ) return result + except OperationValidationError as e: + logger.warning(f"Operation rejected: {e}") + return {"error": str(e)} except Exception as e: logger.error(f"Error listing tags: {e}", exc_info=True) return {"error": str(e)} @@ -2341,6 +2426,9 @@ def _register_get_bank(mcp: FastMCP, memory: MemoryEngine, config: MCPToolsConfi if "disposition" in profile and hasattr(profile["disposition"], "model_dump"): profile["disposition"] = profile["disposition"].model_dump() return json.dumps(profile, indent=2, default=str) + except OperationValidationError as e: + logger.warning(f"Operation rejected: {e}") + return json.dumps({"error": str(e)}) except Exception as e: logger.error(f"Error getting bank: {e}", exc_info=True) return f'{{"error": "{e}"}}' @@ -2366,6 +2454,9 @@ def _register_get_bank(mcp: FastMCP, memory: MemoryEngine, config: MCPToolsConfi if "disposition" in profile and hasattr(profile["disposition"], "model_dump"): profile["disposition"] = profile["disposition"].model_dump() return profile + except OperationValidationError as e: + logger.warning(f"Operation rejected: {e}") + return {"error": str(e)} except Exception as e: logger.error(f"Error getting bank: {e}", exc_info=True) return {"error": str(e)} @@ -2396,6 +2487,9 @@ def _register_get_bank_stats(mcp: FastMCP, memory: MemoryEngine, config: MCPTool request_context=_get_request_context(config), ) return json.dumps(result, indent=2, default=str) + except OperationValidationError as e: + logger.warning(f"Operation rejected: {e}") + return json.dumps({"error": str(e)}) except Exception as e: logger.error(f"Error getting bank stats: {e}", exc_info=True) return f'{{"error": "{e}"}}' @@ -2434,6 +2528,9 @@ def _register_update_bank(mcp: FastMCP, memory: MemoryEngine, config: MCPToolsCo request_context=_get_request_context(config), ) return json.dumps(result, indent=2, default=str) + except OperationValidationError as e: + logger.warning(f"Operation rejected: {e}") + return json.dumps({"error": str(e)}) except Exception as e: logger.error(f"Error updating bank: {e}", exc_info=True) return f'{{"error": "{e}"}}' @@ -2466,6 +2563,9 @@ def _register_update_bank(mcp: FastMCP, memory: MemoryEngine, config: MCPToolsCo request_context=_get_request_context(config), ) return result + except OperationValidationError as e: + logger.warning(f"Operation rejected: {e}") + return {"error": str(e)} except Exception as e: logger.error(f"Error updating bank: {e}", exc_info=True) return {"error": str(e)} @@ -2499,6 +2599,9 @@ def _register_delete_bank(mcp: FastMCP, memory: MemoryEngine, config: MCPToolsCo request_context=_get_request_context(config), ) return json.dumps({"status": "deleted", "bank_id": target_bank, **result}, default=str) + except OperationValidationError as e: + logger.warning(f"Operation rejected: {e}") + return json.dumps({"error": str(e)}) except Exception as e: logger.error(f"Error deleting bank: {e}", exc_info=True) return f'{{"error": "{e}"}}' @@ -2523,6 +2626,9 @@ def _register_delete_bank(mcp: FastMCP, memory: MemoryEngine, config: MCPToolsCo request_context=_get_request_context(config), ) return {"status": "deleted", "bank_id": target_bank, **result} + except OperationValidationError as e: + logger.warning(f"Operation rejected: {e}") + return {"error": str(e)} except Exception as e: logger.error(f"Error deleting bank: {e}", exc_info=True) return {"error": str(e)} @@ -2558,6 +2664,9 @@ def _register_clear_memories(mcp: FastMCP, memory: MemoryEngine, config: MCPTool request_context=_get_request_context(config), ) return json.dumps({"status": "cleared", "bank_id": target_bank, **result}, default=str) + except OperationValidationError as e: + logger.warning(f"Operation rejected: {e}") + return json.dumps({"error": str(e)}) except Exception as e: logger.error(f"Error clearing memories: {e}", exc_info=True) return f'{{"error": "{e}"}}' @@ -2587,6 +2696,9 @@ def _register_clear_memories(mcp: FastMCP, memory: MemoryEngine, config: MCPTool request_context=_get_request_context(config), ) return {"status": "cleared", "bank_id": target_bank, **result} + except OperationValidationError as e: + logger.warning(f"Operation rejected: {e}") + return {"error": str(e)} except Exception as e: logger.error(f"Error clearing memories: {e}", exc_info=True) return {"error": str(e)} diff --git a/hindsight-api/hindsight_api/models.py b/hindsight-api/hindsight_api/models.py index 38b1a4e5..fd20fc0f 100644 --- a/hindsight-api/hindsight_api/models.py +++ b/hindsight-api/hindsight_api/models.py @@ -22,6 +22,7 @@ class RequestContext: tenant_id: str | None = None # Tenant identifier (set by extension after auth) internal: bool = False # True for background/internal operations (skips extension auth) user_initiated: bool = False # True for async operations that originated from a user request + allowed_bank_ids: list[str] | None = None # None = unrestricted (all banks) from pgvector.sqlalchemy import Vector diff --git a/hindsight-api/tests/test_async_retain_tags.py b/hindsight-api/tests/test_async_retain_tags.py index 6804cc33..83c6e2bc 100644 --- a/hindsight-api/tests/test_async_retain_tags.py +++ b/hindsight-api/tests/test_async_retain_tags.py @@ -14,6 +14,7 @@ async def test_submit_async_retain_includes_document_tags_in_task_payload(): engine = MemoryEngine.__new__(MemoryEngine) engine._initialized = True engine._authenticate_tenant = AsyncMock() + engine._operation_validator = None engine._submit_async_operation = AsyncMock(return_value={"operation_id": "op-1"}) # Mock the pool and connection for parent operation creation