feat: 实现 Agent Query Sink

.changes/+agent-query-sink.added.md

@@ -0,0 +1 @@
+Add the asynchronous Agent Query Sink, owner-local Agent read tools, and Sink type discovery.

app/business/agent/AGENTS.md

@@ -14,6 +14,9 @@ transport policy.
   shape is a static type-checking contract, not a second runtime schema.
 - Tool registration is decorator-owned. Exact persisted Tool IDs have set semantics and are bound once per new Thread;
   later registry changes do not rewrite existing Thread schemas or handlers.
+- Tool handlers are owner-local controllers: they own Agent input/result projection and call ordinary domain services.
+  Do not move domain behavior into this subtree or create a central `agent_tools` business package. The durable boundary is
+  documented in `docs/30-unit-tdd/business-pipeline-and-authority.md`.
 - Cancellation owns no rollback, retry, shielding, or compensation. Completed Tool effects remain.
 - `OBSRV__AGENT_DEBUG` enables development events through existing logging. These are diagnostic records, not execution
   persistence or recovery authority. Preserve the actual ToolResult and Turn outcome when changing debug instrumentation.

app/business/agent/__init__.py

@@ -11,6 +11,7 @@
   ToolExecutionError,
 )
 from .main import AgentManager
+from .bootstrap import register_core_agent_tools
 from .persistence import (
   InMemoryThreadPersistenceBackend,
   ThreadID,
@@ -22,6 +23,7 @@
 __all__ = [
   "AgentError",
   "AgentManager",
+  "register_core_agent_tools",
   "AgentNotFoundError",
   "AgentToolBindingError",
   "AgentTurnActiveError",

app/business/agent/bootstrap.py

@@ -0,0 +1,17 @@
+"""Explicit registration entrypoint for core-owned Agent Tool controllers."""
+
+
+def register_core_agent_tools() -> None:
+  """Import every core Tool owner, including Sink delivery controllers.
+
+  Decorators register Tools when their modules load; Python's import cache makes
+  repeated bootstrap calls harmless. Entry points must call this explicitly,
+  rather than rely on routes or package re-exports importing the owners first.
+  """
+  from app.business.graph_navigation_retrieval import tools as _graph_tools
+  from app.business.info_base import tools as _info_base_tools
+  from app.business.info_base.resolver import tools as _resolver_tools
+  from app.business.organization import tools as _organization_tools
+  from app.business.sink import agent_query as _agent_query
+
+  del _graph_tools, _info_base_tools, _resolver_tools, _organization_tools, _agent_query

app/business/agent/projection.py

@@ -0,0 +1,45 @@
+"""JSON projection shared by domain-owned Agent Tool controllers."""
+
+import dataclasses
+import typing
+
+import pydantic
+
+from app.schemas.ai import JSONValue
+
+
+_JSON_ADAPTER = pydantic.TypeAdapter(JSONValue)
+
+
+def project_json(value: typing.Any) -> JSONValue:
+  """Project ordinary domain values without admitting binary Tool results."""
+  if _contains_bytes(value):
+    raise TypeError("Binary values are unavailable through JSON Agent Tools")
+  if isinstance(value, pydantic.BaseModel):
+    projected = value.model_dump(mode="json")
+  elif dataclasses.is_dataclass(value) and not isinstance(value, type):
+    projected = dataclasses.asdict(value)
+  else:
+    projected = pydantic.TypeAdapter(typing.Any).dump_python(
+      typing.cast(typing.Any, value),
+      mode="json",
+    )
+  return _JSON_ADAPTER.validate_python(projected)
+
+
+def _contains_bytes(value: typing.Any) -> bool:
+  if isinstance(value, bytes):
+    return True
+  if isinstance(value, pydantic.BaseModel):
+    return any(
+      _contains_bytes(getattr(value, field)) for field in value.__class__.model_fields
+    )
+  if dataclasses.is_dataclass(value) and not isinstance(value, type):
+    return any(
+      _contains_bytes(getattr(value, field.name)) for field in dataclasses.fields(value)
+    )
+  if isinstance(value, dict):
+    return any(_contains_bytes(item) for item in value.values())
+  if isinstance(value, list | tuple | set | frozenset):
+    return any(_contains_bytes(item) for item in value)
+  return False

app/business/graph_navigation_retrieval/tools.py

@@ -0,0 +1,148 @@
+"""Agent Tool controllers for graph-navigation retrieval."""
+
+import typing
+
+import pydantic
+
+from app.business.agent import AgentManager, ToolExecutionError
+from app.business.agent.projection import project_json
+from app.schemas.ai import JSONValue
+from app.schemas.graph_navigation_retrieval import (
+  DEFAULT_MAX_EXPLORED_BLOCKS,
+  DEFAULT_MAX_EXPLORED_RELATIONS,
+  DEFAULT_MAX_HOPS,
+  DEFAULT_NEIGHBORHOOD_LIMIT,
+  MAX_MAX_EXPLORED_BLOCKS,
+  MAX_MAX_HOPS,
+  MAX_NEIGHBORHOOD_LIMIT,
+  GraphDirection,
+)
+from app.schemas.info_base.block import BlockID
+from app.schemas.info_base.relation import RelationID
+
+from .main import GraphNavigationRetrievalManager
+
+
+GET_ENTITY_NEIGHBORHOOD_TOOL = "get_entity_neighborhood"
+FIND_PATH_TOOL = "find_path"
+GET_CONNECTED_COMPONENTS_TOOL = "get_connected_components"
+
+
+class BlockNeighborhoodInput(pydantic.BaseModel):
+  model_config = pydantic.ConfigDict(extra="forbid", frozen=True)
+
+  entity_type: typing.Literal["block"]
+  entity_id: BlockID
+  direction: GraphDirection = "both"
+  contents: tuple[str, ...] = ()
+  limit: int = pydantic.Field(
+    default=DEFAULT_NEIGHBORHOOD_LIMIT, ge=1, le=MAX_NEIGHBORHOOD_LIMIT
+  )
+  cursor: RelationID | None = None
+
+
+class RelationNeighborhoodInput(pydantic.BaseModel):
+  model_config = pydantic.ConfigDict(extra="forbid", frozen=True)
+
+  entity_type: typing.Literal["relation"]
+  entity_id: RelationID
+
+
+class EntityNeighborhoodInput(
+  pydantic.RootModel[
+    typing.Annotated[
+      BlockNeighborhoodInput | RelationNeighborhoodInput,
+      pydantic.Field(discriminator="entity_type"),
+    ]
+  ]
+):
+  model_config = pydantic.ConfigDict(json_schema_extra={"type": "object"})
+
+  @classmethod
+  def model_json_schema(cls, *args, **kwargs) -> dict[str, typing.Any]:
+    schema = super().model_json_schema(*args, **kwargs)
+    properties = BlockNeighborhoodInput.model_json_schema(*args, **kwargs)["properties"]
+    properties["entity_type"] = {"type": "string", "enum": ["block", "relation"]}
+    schema["properties"] = properties
+    return schema
+
+
+class FindPathInput(pydantic.BaseModel):
+  model_config = pydantic.ConfigDict(extra="forbid", frozen=True)
+
+  from_block_id: BlockID
+  to_block_id: BlockID
+  direction: GraphDirection = "both"
+  contents: tuple[str, ...] = ()
+  max_hops: int = pydantic.Field(default=DEFAULT_MAX_HOPS, ge=0, le=MAX_MAX_HOPS)
+  max_explored_blocks: int = pydantic.Field(
+    default=DEFAULT_MAX_EXPLORED_BLOCKS, ge=1, le=MAX_MAX_EXPLORED_BLOCKS
+  )
+
+
+class ConnectedComponentsInput(pydantic.BaseModel):
+  model_config = pydantic.ConfigDict(extra="forbid", frozen=True)
+
+  seed_block_ids: tuple[BlockID, ...]
+  contents: tuple[str, ...] = pydantic.Field(min_length=1)
+  max_explored_blocks: int = pydantic.Field(default=DEFAULT_MAX_EXPLORED_BLOCKS, ge=1)
+  max_explored_relations: int = pydantic.Field(default=DEFAULT_MAX_EXPLORED_RELATIONS, ge=1)
+
+
+@AgentManager.tool(
+  GET_ENTITY_NEIGHBORHOOD_TOOL,
+  description=(
+    "Read a Block's direct neighborhood or a Relation with its endpoints. "
+    "Null may indicate an incorrect entity type."
+  ),
+)
+async def get_entity_neighborhood(input: EntityNeighborhoodInput) -> JSONValue:
+  request = input.root
+  if request.entity_type == "block":
+    result = await GraphNavigationRetrievalManager.get_block_neighborhood(
+      request.entity_id,
+      direction=request.direction,
+      contents=request.contents,
+      limit=request.limit,
+      cursor=request.cursor,
+    )
+  else:
+    result = await GraphNavigationRetrievalManager.get_relation_neighborhood(
+      request.entity_id
+    )
+  return project_json(result)
+
+
+@AgentManager.tool(
+  FIND_PATH_TOOL,
+  description="Find a bounded graph path; an exploration limit is not proof of absence.",
+)
+async def find_path(input: FindPathInput) -> JSONValue:
+  return project_json(
+    await GraphNavigationRetrievalManager.find_path(
+      input.from_block_id,
+      input.to_block_id,
+      direction=input.direction,
+      contents=input.contents,
+      max_hops=input.max_hops,
+      max_explored_blocks=input.max_explored_blocks,
+    )
+  )
+
+
+@AgentManager.tool(
+  GET_CONNECTED_COMPONENTS_TOOL,
+  description=(
+    "Partition seeds by bounded undirected reachability through exact Relation contents."
+  ),
+)
+async def get_connected_components(input: ConnectedComponentsInput) -> JSONValue:
+  try:
+    result = await GraphNavigationRetrievalManager.get_connected_components(
+      **input.model_dump()
+    )
+  except ValueError as error:
+    raise ToolExecutionError(
+      {"error": type(error).__name__, "message": str(error)}
+    ) from error
+  return project_json(result)

app/business/info_base/main.py

@@ -1,4 +1,16 @@
-"""Info-base graph command coordination."""
+"""Info-base graph command coordination and cross-retrieval use cases."""
+
+import asyncio
+from dataclasses import dataclass
+import typing
+
+from app.business.info_base.services import BlockService, get_entity_records
+from app.business.lexical_retrieval import LexicalRetrievalManager
+from app.business.semantic_retrieval import SemanticRetrievalManager
+from app.schemas.lexical_retrieval import LexicalRetrievalResult
+from app.schemas.semantic_retrieval import SemanticRetrievalResult, VectorRetrievalOptions
+from app.schemas.info_base.block import BlockModel
+from app.schemas.info_base.relation import RelationModel
 
 from app.schemas.info_base.main import (
   GraphBlockForm,
@@ -8,8 +20,60 @@
 )
 
 
+@dataclass(frozen=True)
+class RetrievalBranch:
+  result: LexicalRetrievalResult | SemanticRetrievalResult | None = None
+  error: BaseException | None = None
+
+
 class InfoBaseManager:
-  """Own graph-form normalization and graph insertion coordination."""
+  """Own graph commands and use cases that span info-base read capabilities."""
+
+  @classmethod
+  async def retrieve(
+    cls,
+    query: str,
+    mode: typing.Literal["lexical", "semantic", "hybrid"] = "hybrid",
+    limit: int = 20,
+  ) -> dict[str, RetrievalBranch]:
+    """Run selected retrieval owners independently without merging their ranking."""
+    operations: dict[str, typing.Awaitable] = {}
+    if mode in {"lexical", "hybrid"}:
+      operations["lexical"] = LexicalRetrievalManager.retrieve_local(query, limit)
+    if mode in {"semantic", "hybrid"}:
+      operations["semantic"] = SemanticRetrievalManager.retrieve_local(
+        query, options=VectorRetrievalOptions(limit=limit)
+      )
+    outcomes = await asyncio.gather(*operations.values(), return_exceptions=True)
+    return {
+      name: RetrievalBranch(
+        error=outcome if isinstance(outcome, BaseException) else None,
+        result=None if isinstance(outcome, BaseException) else outcome,
+      )
+      for name, outcome in zip(operations, outcomes, strict=True)
+    }
+
+  @classmethod
+  async def get_entities(
+    cls,
+    entities: typing.Collection[tuple[typing.Literal["block", "relation"], int]],
+    *,
+    random_count: int = 1,
+  ) -> tuple[BlockModel | RelationModel | None, ...]:
+    """Read ordered persisted entities, or random Blocks when selection is empty."""
+    if not entities:
+      return typing.cast(
+        tuple[BlockModel | RelationModel | None, ...],
+        await BlockService.get_random_many(random_count),
+      )
+    blocks, relations = await get_entity_records(
+      tuple(identity for kind, identity in entities if kind == "block"),
+      tuple(identity for kind, identity in entities if kind == "relation"),
+    )
+    return tuple(
+      blocks.get(identity) if kind == "block" else relations.get(identity)
+      for kind, identity in entities
+    )
 
   @classmethod
   def normalize_graph(