fix(api): GET /banks/{bank_id}/profile no longer auto-creates the bank (#1287)

* fix(api): make GET /banks/{bank_id}/profile a true read (no auto-create)

The HTTP GET handler for bank profile was calling
get_or_create_bank_profile, so a request for a non-existent bank would
silently create it as a side effect. This is dangerous for any client
that polls or holds a stale bank_id while the surrounding context
(tenant, schema, user session) changes — the GET would create the
bank in whatever tenant the request was authenticated against, not
the tenant the client originally meant.

Reads must not have create-as-side-effect. Changes:

* Add bank_utils.get_bank_profile_if_exists(pool, bank_id) — pure
  read; returns None when the row is absent.
* memory_engine.get_bank_profile gets a create_if_missing kwarg
  (defaults True for backwards compatibility). When False, uses the
  new pure-read path and returns None on miss; the caller is
  responsible for translating None to a 404.
* Read-only HTTP endpoints pass create_if_missing=False:
  - GET /v1/default/banks/{bank_id}/profile
  - GET /v1/default/banks/{bank_id}/template (export)
  - GET /v1/default/banks/{bank_id}/audit/logs
  - GET /v1/default/banks/{bank_id}/audit/stats
  All four now return 404 for a missing bank instead of silently
  materializing one.
* Write paths (PUT/PATCH bank, import template, MCP retain/recall)
  keep the default create_if_missing=True — they have explicit
  expectations about creating banks on first use.

Test: tests/test_agents_api.py adds
test_get_bank_profile_no_auto_create_returns_none asserting that a
missing bank is not created as a side effect of a read, and that
explicit auto-create still works after.

* chore(api): @overload get_bank_profile so existing callers stay non-Optional

The previous commit added a create_if_missing kwarg to get_bank_profile
and changed the return annotation to dict[str, Any] | None. That made
the type checker treat every existing caller as receiving Optional,
producing 12 not-subscriptable errors in mcp_tools.py where callers
assumed non-None.

Add @overload variants so the precise return type is recovered:
  - create_if_missing=Literal[True] (the default)  -> dict[str, Any]
  - create_if_missing=Literal[False] (explicit)    -> dict[str, Any] | None

The interface.py abstract declaration mirrors the new signature.
ty check hindsight_api/ is clean after this change.
This commit is contained in:
Chris Bartholomew
2026-04-28 11:13:22 +02:00
committed by GitHub
parent 91106f30ef
commit 99a8978905
5 changed files with 149 additions and 14 deletions
+30 -7
View File
@@ -4666,7 +4666,7 @@ def _register_routes(app: FastAPI):
"/v1/default/banks/{bank_id}/profile",
response_model=BankProfileResponse,
summary="Get memory bank profile",
description="Get disposition traits and mission for a memory bank. Auto-creates agent with defaults if not exists.",
description="Get disposition traits and mission for a memory bank. Returns 404 if the bank does not exist.",
operation_id="get_bank_profile",
tags=["Banks"],
deprecated=True,
@@ -4674,7 +4674,15 @@ def _register_routes(app: FastAPI):
async def api_get_bank_profile(bank_id: str, request_context: RequestContext = Depends(get_request_context)):
"""Get memory bank profile (disposition + mission)."""
try:
profile = await app.state.memory.get_bank_profile(bank_id, request_context=request_context)
# Read endpoints must not have create-as-side-effect: a client
# holding onto a stale bank_id (e.g., a UI polling after the user
# changed context) would otherwise silently re-create the bank in
# an unrelated tenant. Surface a missing bank as 404.
profile = await app.state.memory.get_bank_profile(
bank_id, request_context=request_context, create_if_missing=False
)
if profile is None:
raise HTTPException(status_code=404, detail=f"Bank '{bank_id}' not found")
# Convert DispositionTraits object to dict for Pydantic
disposition_dict = (
profile["disposition"].model_dump()
@@ -5010,8 +5018,10 @@ def _register_routes(app: FastAPI):
):
"""Export a bank's config and mental models as a template manifest."""
try:
# Authenticate and ensure bank exists
profile = await app.state.memory.get_bank_profile(bank_id, request_context=request_context)
# Read endpoint: do not auto-create on missing bank.
profile = await app.state.memory.get_bank_profile(
bank_id, request_context=request_context, create_if_missing=False
)
if profile is None:
raise HTTPException(status_code=404, detail=f"Bank '{bank_id}' not found")
@@ -6110,8 +6120,14 @@ def _register_routes(app: FastAPI):
pool = await app.state.memory._get_pool()
# Ensure bank exists
await app.state.memory.get_bank_profile(bank_id, request_context=request_context)
# Read endpoint: verify bank exists without auto-creating it.
if (
await app.state.memory.get_bank_profile(
bank_id, request_context=request_context, create_if_missing=False
)
is None
):
raise HTTPException(status_code=404, detail=f"Bank '{bank_id}' not found")
from hindsight_api.engine.db_utils import acquire_with_retry
@@ -6225,7 +6241,14 @@ def _register_routes(app: FastAPI):
from hindsight_api.engine.memory_engine import fq_table
pool = await app.state.memory._get_pool()
await app.state.memory.get_bank_profile(bank_id, request_context=request_context)
# Read endpoint: verify bank exists without auto-creating it.
if (
await app.state.memory.get_bank_profile(
bank_id, request_context=request_context, create_if_missing=False
)
is None
):
raise HTTPException(status_code=404, detail=f"Bank '{bank_id}' not found")
# Determine time range (always per-day buckets)
from datetime import timedelta as _td
@@ -161,16 +161,22 @@ class MemoryEngineInterface(ABC):
bank_id: str,
*,
request_context: "RequestContext",
) -> dict[str, Any]:
create_if_missing: bool = True,
) -> dict[str, Any] | None:
"""
Get bank profile including disposition and mission.
Args:
bank_id: The memory bank ID.
request_context: Request context for authentication.
create_if_missing: If True (default), the bank is auto-created
with defaults if it does not exist. Pass False to make this
a strict read — returns None if the bank does not exist.
Returns:
Bank profile dict with bank_id, name, disposition, and mission.
Bank profile dict with bank_id, name, disposition, and mission,
or None when create_if_missing=False and the bank does not
exist.
"""
...
@@ -18,7 +18,7 @@ import uuid
from collections.abc import Awaitable, Callable
from dataclasses import dataclass
from datetime import UTC, datetime, timedelta, timezone
from typing import TYPE_CHECKING, Any
from typing import TYPE_CHECKING, Any, Literal, overload
import asyncpg
import httpx
@@ -5476,22 +5476,51 @@ class MemoryEngine(MemoryEngineInterface):
# ==================== bank profile Methods ====================
# Type-checker overloads: when create_if_missing is True (the default),
# this method always returns a profile dict — the type checker can rely
# on non-None for every existing caller. Only when create_if_missing is
# explicitly False does the return become Optional.
@overload
async def get_bank_profile(
self,
bank_id: str,
*,
request_context: "RequestContext",
) -> dict[str, Any]:
create_if_missing: Literal[True] = True,
) -> dict[str, Any]: ...
@overload
async def get_bank_profile(
self,
bank_id: str,
*,
request_context: "RequestContext",
create_if_missing: Literal[False],
) -> dict[str, Any] | None: ...
async def get_bank_profile(
self,
bank_id: str,
*,
request_context: "RequestContext",
create_if_missing: bool = True,
) -> dict[str, Any] | None:
"""
Get bank profile (name, disposition + mission).
Auto-creates agent with default values if not exists.
Args:
bank_id: bank IDentifier
request_context: Request context for authentication.
create_if_missing: If True (default), the bank is auto-created
with defaults when it does not exist. Pass False from read-
only callers (HTTP GET handlers, polling, etc.) so a missing
bank surfaces as None rather than being silently created.
The caller is then responsible for translating None to a
404 (or similar).
Returns:
Dict with name, disposition traits, and mission
Dict with name, disposition traits, and mission, or None when
create_if_missing=False and the bank does not exist.
"""
await self._authenticate_tenant(request_context)
if self._operation_validator:
@@ -5500,7 +5529,13 @@ class MemoryEngine(MemoryEngineInterface):
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, created = await bank_utils.get_or_create_bank_profile(pool, bank_id)
if not create_if_missing:
existing = await bank_utils.get_bank_profile_if_exists(pool, bank_id)
if existing is None:
return None
profile, created = existing, False
else:
profile, created = await bank_utils.get_or_create_bank_profile(pool, bank_id)
# Apply HINDSIGHT_API_DEFAULT_BANK_TEMPLATE to freshly-created banks. Done
# before reading the resolved config below so the template's overrides
@@ -117,6 +117,41 @@ async def get_bank_profile(pool, bank_id: str) -> BankProfile:
return profile
async def get_bank_profile_if_exists(pool, bank_id: str) -> BankProfile | None:
"""
Get bank profile (name, disposition + mission) without auto-creating.
Returns None if the bank does not exist. This is the read-only variant
of get_bank_profile, intended for read endpoints where a bank that
doesn't exist should surface as 404 rather than be silently created.
Args:
pool: Database connection pool
bank_id: bank IDentifier
Returns:
BankProfile if the bank exists, otherwise None.
"""
async with acquire_with_retry(pool) as conn:
row = await conn.fetchrow(
f"""
SELECT name, disposition, mission
FROM {fq_table("banks")} WHERE bank_id = $1
""",
bank_id,
)
if not row:
return None
disposition_data = row["disposition"]
if isinstance(disposition_data, str):
disposition_data = json.loads(disposition_data)
return BankProfile(
name=row["name"],
disposition=DispositionTraits(**disposition_data),
mission=row["mission"] or "",
)
async def get_or_create_bank_profile(pool, bank_id: str) -> tuple[BankProfile, bool]:
"""
Get bank profile, auto-creating with defaults if it doesn't exist.
@@ -16,6 +16,42 @@ def unique_agent_id(prefix: str) -> str:
class TestAgentProfile:
"""Tests for agent profile management."""
@pytest.mark.asyncio
async def test_get_bank_profile_no_auto_create_returns_none(
self, memory: MemoryEngine, request_context
):
"""When create_if_missing=False is passed, a missing bank returns None
rather than being silently auto-created. This is what read-only
endpoints (HTTP GET, polling, etc.) must use to avoid creating banks
as a side effect of a stale client request."""
bank_id = unique_agent_id("test_no_auto_create")
# First call with create_if_missing=False on a non-existent bank
result = await memory.get_bank_profile(
bank_id, request_context=request_context, create_if_missing=False
)
assert result is None, "Expected None for missing bank with create_if_missing=False"
# Verify the bank was NOT created as a side effect
result_again = await memory.get_bank_profile(
bank_id, request_context=request_context, create_if_missing=False
)
assert result_again is None, "Bank must not exist after read-only call"
# And explicit auto-create still works
created = await memory.get_bank_profile(
bank_id, request_context=request_context, create_if_missing=True
)
assert created is not None
assert created["disposition"]["skepticism"] == 3
# Now read-only call sees it
seen = await memory.get_bank_profile(
bank_id, request_context=request_context, create_if_missing=False
)
assert seen is not None
assert seen["disposition"]["skepticism"] == 3
@pytest.mark.asyncio
async def test_get_agent_profile_creates_default(self, memory: MemoryEngine, request_context):
"""Test that getting a profile for a new agent creates default disposition."""