diff --git a/.bumpversion.toml b/.bumpversion.toml index f5825d4e..fade34b9 100644 --- a/.bumpversion.toml +++ b/.bumpversion.toml @@ -3,7 +3,7 @@ # https://peps.python.org/pep-0440/ [tool.bumpversion] -current_version = "1.0.3" +current_version = "1.0.4.dev0" parse = """(?x) (?P0|[1-9]\\d*)\\. (?P0|[1-9]\\d*)\\. diff --git a/pyproject.toml b/pyproject.toml index 521a0684..af85808a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -41,7 +41,7 @@ "pydantic>=2.12.4", "redis[hiredis]>=7.4.0,<9", ] - version = "1.0.3" + version = "1.0.4.dev0" [project.optional-dependencies] agno = [ "agno>=2.6" ] diff --git a/src/digitalkin/__version__.py b/src/digitalkin/__version__.py index f1c10026..7db13246 100644 --- a/src/digitalkin/__version__.py +++ b/src/digitalkin/__version__.py @@ -5,4 +5,4 @@ try: __version__ = version("digitalkin") except PackageNotFoundError: - __version__ = "1.0.3" + __version__ = "1.0.4.dev0" diff --git a/src/digitalkin/community/agno/agno_adapter.py b/src/digitalkin/community/agno/agno_adapter.py index 5f3e4881..eb5997e4 100644 --- a/src/digitalkin/community/agno/agno_adapter.py +++ b/src/digitalkin/community/agno/agno_adapter.py @@ -4,6 +4,7 @@ import json import logging +import time import uuid from typing import TYPE_CHECKING, Any, TypeAlias @@ -99,8 +100,9 @@ def __init__(self) -> None: self._completed_run_ids: set[str] = set() # Delegations in flight, keyed by the child's run id — which is exactly the # ``subagent_run_id`` every event that child produces carries. Value is its display - # name and the metadata of the event that opened it. - self._subagents: dict[str, tuple[str, dict[str, Any] | None]] = {} + # name, the metadata of the event that opened it, and the ``time.monotonic()`` + # timestamp it started at (used to log the member's run duration on completion). + self._subagents: dict[str, tuple[str, dict[str, Any] | None, float]] = {} self._is_paused: bool = False self._paused_tool_executions: list[Any] = [] @@ -269,7 +271,7 @@ def _handle_run_started(self, agno_event: AgnoRunStartedEvent, timestamp: Any) - # named step so the client can show progress, and close the parent's open # bubble so the member's content lands in its own labelled message. if not run_id: - logger.info("[agno-adapter] DROP nested run_started without run_id parent_run_id=%s", parent_run_id) + logger.debug("[agno-adapter] DROP nested run_started without run_id parent_run_id=%s", parent_run_id) return [] # Close the parent's own sequences so its bubble does not stay open across the @@ -281,7 +283,7 @@ def _handle_run_started(self, agno_event: AgnoRunStartedEvent, timestamp: Any) - # Attribution is by id, so the name is a plain label — no need to disambiguate # members that share one, as the step-based scheme required. name = (self._last_metadata or {}).get("name") or "member" - self._subagents[run_id] = (name, self._last_metadata) + self._subagents[run_id] = (name, self._last_metadata, time.monotonic()) logger.info( "[agno-adapter] SUBAGENT_STARTED name=%s subagent_run_id=%s parent_run_id=%s", @@ -303,10 +305,10 @@ def _handle_run_started(self, agno_event: AgnoRunStartedEvent, timestamp: Any) - return events if run_id and run_id == self._active_run_id: - logger.info("[agno-adapter] DROP duplicate run_started run_id=%s", run_id) + logger.debug("[agno-adapter] DROP duplicate run_started run_id=%s", run_id) return [] - logger.info( + logger.debug( "[agno-adapter] EMIT run_started run_id=%s session_id=%s active_was=%s metadata=%s", run_id, getattr(agno_event, "session_id", None), @@ -356,20 +358,31 @@ def _handle_run_completed(self, agno_event: AgnoRunCompletedEvent, timestamp: An metadata=subagent[1], ) ) - logger.info( - "[agno-adapter] SUBAGENT_FINISHED name=%s subagent_run_id=%s parent_run_id=%s closed=%d", - subagent[0] if subagent else None, - run_id, - parent_run_id, - len(events), - ) + if subagent is None: + logger.info( + "[agno-adapter] SUBAGENT_FINISHED name=unknown subagent_run_id=%s parent_run_id=%s closed=%d", + run_id, + parent_run_id, + len(events), + ) + else: + elapsed_s = time.monotonic() - subagent[2] + logger.info( + "[agno-adapter] SUBAGENT_FINISHED name=%s subagent_run_id=%s parent_run_id=%s " + "closed=%d elapsed_s=%.1f", + subagent[0], + run_id, + parent_run_id, + len(events), + elapsed_s, + ) return events if run_id and run_id in self._completed_run_ids and run_id != self._active_run_id: - logger.info("[agno-adapter] DROP duplicate run_completed run_id=%s", run_id) + logger.debug("[agno-adapter] DROP duplicate run_completed run_id=%s", run_id) return [] - logger.info( + logger.debug( "[agno-adapter] EMIT run_completed run_id=%s active_run_id=%s", run_id, self._active_run_id, @@ -414,7 +427,13 @@ def _handle_run_error(self, agno_event: AgnoRunErrorEvent, timestamp: Any) -> li events = self._close_content(run_id, timestamp) events.extend(self._close_reasoning(run_id, timestamp)) subagent = self._subagents.pop(run_id) - logger.info("[agno-adapter] SUBAGENT_ERROR name=%s subagent_run_id=%s", subagent[0], run_id) + elapsed_s = time.monotonic() - subagent[2] + logger.info( + "[agno-adapter] SUBAGENT_ERROR name=%s subagent_run_id=%s elapsed_s=%.1f", + subagent[0], + run_id, + elapsed_s, + ) events.append( SubagentErrorEvent( event=AgentRunEvent.SUBAGENT_ERROR, @@ -839,7 +858,7 @@ def _close_subagents(self, timestamp: Any) -> list[BaseAgentRunEvent]: timestamp=timestamp, metadata=metadata, ) - for run_id, (_, metadata) in reversed(list(self._subagents.items())) + for run_id, (_, metadata, _start) in reversed(list(self._subagents.items())) ] self._subagents.clear() return events diff --git a/src/digitalkin/community/agno/module_toolkit.py b/src/digitalkin/community/agno/module_toolkit.py index 31114384..0e7ca156 100644 --- a/src/digitalkin/community/agno/module_toolkit.py +++ b/src/digitalkin/community/agno/module_toolkit.py @@ -24,6 +24,7 @@ from digitalkin.logger import logger from digitalkin.models.module import ModuleContext from digitalkin.models.module.ag_ui import AgUiCustomEventOutput, AgUiOutput +from digitalkin.models.module.module_context import STREAM_SENTINEL_PROTOCOLS from digitalkin.models.module.tool_cache import ToolDefinition, ToolModuleInfo # Default timeout for tool calls in seconds @@ -89,30 +90,11 @@ def __init__( self._tool_module_info = tool_module_info self._timeout = timeout_seconds - sdk_tool_names = sorted(t.name for t in tool_module_info.tools) - logger.info( - "Creating ModuleToolkit: setup_id='%s' slug='%s' module_name='%s' sdk_tools_count=%d sdk_tool_names=%s", - tool_module_info.setup_id, - tool_module_info.slug, - tool_module_info.module_name, - len(tool_module_info.tools), - sdk_tool_names, - ) - tool_functions = context.create_tool_functions(tool_module_info.setup_id) if allowed_tools is not None: tool_functions = [(td, fn) for td, fn in tool_functions if td.name in allowed_tools] - fn_names = sorted(td.name for td, _ in tool_functions) - logger.info( - "Built tool_functions: setup_id='%s' slug='%s' fn_count=%d fn_names=%s", - tool_module_info.setup_id, - tool_module_info.slug, - len(tool_functions), - fn_names, - ) - # Function objects with explicit JSON schema + skip_entrypoint_processing=True # bypass Agno's inspect.signature() introspection, which sees **kwargs: Any and # generates {kwargs: object} — causing the LLM to miss required parameters. @@ -129,16 +111,9 @@ def __init__( skip_entrypoint_processing=True, ) ) - logger.info( - "[lat-audit] tool_wrapped: setup_id='%s' fn_name='%s' param_count=%d param_names=%s desc_chars=%d", - tool_module_info.setup_id, - wrapper.__name__, - tool_def.parameter_count, - sorted(tool_def.parameter_names), - len(tool_def.description or ""), - ) if not agno_functions: + sdk_tool_names = sorted(t.name for t in tool_module_info.tools) if not tool_module_info.tools: reason = "sdk_returned_zero_tools" elif not tool_functions: @@ -156,13 +131,12 @@ def __init__( len(tool_functions), ) - logger.info( - "[lat-audit] toolkit_built: setup_id='%s' slug='%s' sdk_tools=%d wrapped=%d empty=%s", - tool_module_info.setup_id, + logger.debug( + "ModuleToolkit built: slug=%s setup_id=%s tools=%d/%d", tool_module_info.slug, - len(tool_module_info.tools), + tool_module_info.setup_id, len(agno_functions), - not agno_functions, + len(tool_module_info.tools), ) toolkit_name = ( @@ -312,11 +286,12 @@ def _handle_success( if tool_metadata and tool_metadata.cost_estimate_usd is not None: cost_info = f", cost=${tool_metadata.cost_estimate_usd:.4f}" logger.info( - "Tool '%s' completed in %.2fms (success=True%s, images=%d) setup_id=%s task_id=%s", + "Tool '%s' completed in %.2fms (success=True%s, images=%d) args=%s setup_id=%s task_id=%s", tool_name, duration_ms, cost_info, len(image_urls), + sorted(input_kwargs), self._tool_module_info.setup_id, self._context.session.job_id, ) @@ -346,10 +321,11 @@ def _handle_failure( input_kwargs=input_kwargs, ) logger.warning( - "Tool '%s' failed in %.2fms: %s setup_id=%s task_id=%s", + "Tool '%s' failed in %.2fms: %s args=%s setup_id=%s task_id=%s", tool_name, duration_ms, error_msg, + sorted(input_kwargs), self._tool_module_info.setup_id, self._context.session.job_id, ) @@ -362,17 +338,28 @@ def _find_successful_response(results: list[dict[str, Any]]) -> dict[str, Any] | Each response is the ``MessageToDict`` of a payload Struct, shape ``{"root": {"protocol": "...", ...}, "annotations": {...}}``. A "successful" response is the most recent one whose ``root.protocol`` - is *not* a lifecycle/error sentinel. + is *not* a lifecycle/error sentinel. A fatal ``stream.error`` anywhere in + the stream (e.g. a tool that streamed progress lines and then died, or a + gateway idle timeout ending the stream mid-call) means the tool did not + finish, regardless of what streamed before or after it. A non-fatal + ``stream.error`` (the target module kept the stream open) does not fail + the call on its own. Returns: - The matching dict, or None if every response was a sentinel. + The matching dict, or None if any response was a fatal stream.error + or every response was a sentinel. """ + for resp in results: + root = resp.get("root") + if isinstance(root, dict) and root.get("protocol") == "stream.error" and root.get("fatal"): + return None + for resp in reversed(results): root = resp.get("root") if not isinstance(root, dict): continue protocol = root.get("protocol", "") - if protocol in {"stream.start", "stream.end", "stream.init", "stream.error"}: + if protocol in STREAM_SENTINEL_PROTOCOLS: continue return resp return None @@ -477,13 +464,18 @@ def _create_tool_wrapper( tag = f"tool.call[{self._tool_module_info.slug}/{tool_name}]" async def wrapper(**kwargs: Any) -> str | ToolResult: + if context.session.cancelled: + logger.warning("Tool call refused after task cancellation: tool=%s task_id=%s", tool_name, task_id) + msg = f"task cancelled before tool '{tool_name}'" + raise asyncio.CancelledError(msg) + start_time = time.perf_counter() call_timer = StepTimer() outcome = "ok" kwargs = ModuleToolkit._unwrap_kwargs(kwargs, tool_name, expected_params) - logger.info( + logger.debug( "Calling tool '%s' with kwargs: %s setup_id=%s task_id=%s", tool_name, list(kwargs.keys()), diff --git a/src/digitalkin/core/job_manager/single_job_manager.py b/src/digitalkin/core/job_manager/single_job_manager.py index 92c939b8..ebedc231 100644 --- a/src/digitalkin/core/job_manager/single_job_manager.py +++ b/src/digitalkin/core/job_manager/single_job_manager.py @@ -325,7 +325,7 @@ async def run_instance( ) timer.mark("create_task") timer.log("run_instance", task_id=job_id) - logger.info("Managed task started: '%s'", job_id, extra={"task_id": job_id}) + logger.debug("Managed task started: '%s'", job_id, extra={"task_id": job_id}) return job_id async def list_modules(self) -> dict[str, dict[str, Any]]: diff --git a/src/digitalkin/core/task_manager/base_task_manager.py b/src/digitalkin/core/task_manager/base_task_manager.py index b5c9bf62..f0b3a36a 100644 --- a/src/digitalkin/core/task_manager/base_task_manager.py +++ b/src/digitalkin/core/task_manager/base_task_manager.py @@ -102,7 +102,7 @@ async def _cleanup_task(self, task_id: str, mission_id: str) -> None: self._task_slot.release() if get_task_manager_settings().max_queued_tasks > 0: self._system_gate.release() - logger.info( + logger.debug( "Task cleaned up (%d remaining) final_status=%s cancellation_reason=%s", len(self.tasks_sessions), final_status, @@ -194,7 +194,7 @@ async def _acquire_with_queue(self, coro: Coroutine[Any, Any, None]) -> None: self._waiting_count += 1 if self._waiting_count > 0: - logger.info( + logger.debug( "Task queued for execution (%d waiting, %d/%d slots busy)", self._waiting_count, self._active_slots, @@ -302,7 +302,7 @@ async def send_signal(self, task_id: str, mission_id: str, signal_type: str, pay ) return False - logger.info( + logger.debug( "Sending signal '%s' to task '%s'", signal_type, task_id, @@ -425,7 +425,9 @@ async def clean_session(self, task_id: str, mission_id: str) -> bool: else: await self._cleanup_task(task_id, mission_id) - logger.info("Cleaning up session for task: '%s'", task_id, extra={"mission_id": mission_id, "task_id": task_id}) + logger.debug( + "Cleaning up session for task: '%s'", task_id, extra={"mission_id": mission_id, "task_id": task_id} + ) return True async def cancel_all_tasks(self, mission_id: str, timeout: float | None = None) -> dict[str, bool | BaseException]: diff --git a/src/digitalkin/core/task_manager/local_task_manager.py b/src/digitalkin/core/task_manager/local_task_manager.py index 2007a8f3..bc749327 100644 --- a/src/digitalkin/core/task_manager/local_task_manager.py +++ b/src/digitalkin/core/task_manager/local_task_manager.py @@ -55,7 +55,7 @@ async def create_task( session = self._create_session(task_id, mission_id, module) registered = True - logger.info( + logger.debug( "Creating local task: '%s'", task_id, extra={ @@ -77,7 +77,7 @@ async def _finalize() -> None: ) self.tasks[task_id] = supervisor_task - logger.info( + logger.debug( "Local task created and started: '%s' (total_tasks=%d)", task_id, len(self.tasks), diff --git a/src/digitalkin/core/task_manager/module_runner.py b/src/digitalkin/core/task_manager/module_runner.py index 919b6857..2523f679 100644 --- a/src/digitalkin/core/task_manager/module_runner.py +++ b/src/digitalkin/core/task_manager/module_runner.py @@ -136,7 +136,7 @@ async def _on_output(output_data: Any) -> None: await self._redis_client.xadd(stream_key, {"eos": b"true"}) await self._redis_client.expire(stream_key, stream_settings.redis_stream_ttl) t_eos_write_end = time.perf_counter_ns() - logger.info( + logger.debug( "[close-debug] producer_eos_write: xadd_expire=%.2fms t_done_ns=%d task_id=%s", (t_eos_write_end - t_eos_write_start) / 1e6, t_eos_write_end, @@ -165,7 +165,7 @@ async def _on_output(output_data: Any) -> None: top_level_keys = list(query.fields.keys()) query_byte_size = query.ByteSize() - logger.info( + logger.debug( "[input-debug] inbound Struct: top_keys=%s wire_bytes=%d", top_level_keys, query_byte_size, @@ -231,10 +231,12 @@ async def _on_output(output_data: Any) -> None: dict_repr, extra=log_extra, ) - missing_summary = f" missing_fields={missing_paths}" if missing_paths else "" + first_errors = "; ".join( + f"{'.'.join(str(p) for p in e['loc'])}: {e['msg']}" for e in exc.errors(include_url=False)[:3] + )[:500] await on_fatal( StreamErrorCode.INPUT_VALIDATION_ERROR.value, - f"input validation failed for {model_name}: top_keys={top_level_keys}{missing_summary}", + f"input validation failed for {model_name}: {first_errors}", ) except BackpressureTimeoutError as exc: logger.exception("ModuleRunner: backpressure timeout", extra=log_extra) diff --git a/src/digitalkin/core/task_manager/redis/proto_streams.py b/src/digitalkin/core/task_manager/redis/proto_streams.py index cebf8f9c..0673e98d 100644 --- a/src/digitalkin/core/task_manager/redis/proto_streams.py +++ b/src/digitalkin/core/task_manager/redis/proto_streams.py @@ -114,7 +114,7 @@ async def read_structs( # noqa: C901 eos = fields.get(b"eos", b"") if eos == b"true": - logger.info( + logger.debug( "[close-debug] reader_saw_eos: last_xread_block=%.2fms t_seen_ns=%d task_id=%s", (t_xread_end - t_xread_start) / 1e6, t_xread_end, diff --git a/src/digitalkin/core/task_manager/remote_task_manager.py b/src/digitalkin/core/task_manager/remote_task_manager.py index 1e451903..943d9ef5 100644 --- a/src/digitalkin/core/task_manager/remote_task_manager.py +++ b/src/digitalkin/core/task_manager/remote_task_manager.py @@ -46,7 +46,7 @@ async def create_task( self._create_session(task_id, mission_id, module) registered = True - logger.info( + logger.debug( "Registering remote task: '%s'", task_id, extra={ @@ -58,7 +58,7 @@ async def create_task( # Close coroutine - worker will recreate and execute it coro.close() - logger.info( + logger.debug( "Remote task registered: '%s' (total_sessions=%d)", task_id, len(self.tasks_sessions), diff --git a/src/digitalkin/grpc_servers/gateway_servicer.py b/src/digitalkin/grpc_servers/gateway_servicer.py index 52e1324b..f3cad2f9 100644 --- a/src/digitalkin/grpc_servers/gateway_servicer.py +++ b/src/digitalkin/grpc_servers/gateway_servicer.py @@ -271,7 +271,7 @@ async def StartStream( # noqa: PLR0911 return gateway_pb2.StartStreamResponse(accepted=False, task_id=task_id) timer.mark("xadd_stream_start") - logger.info("→ Dial-back scheduled to consumer %s", client_address, extra=log_extra) + logger.debug("→ Dial-back scheduled to consumer %s", client_address, extra=log_extra) self._spawn( self._dial_consumer( task_id=task_id, @@ -625,7 +625,7 @@ async def _consume_from_redis( async for struct_data in reader.read_structs(skip_to_seq=skip_to_seq): if first: t2 = time.perf_counter_ns() - logger.info( + logger.debug( "Stream: cursor=%.1fms xread_wait=%.1fms total_to_first=%.1fms task_id=%s", (t1 - t0) / 1e6, (t2 - t1) / 1e6, @@ -641,7 +641,7 @@ async def _consume_from_redis( seq = reader._last_seq + 2 if resume else seq + 1 # noqa: SLF001 yield self._sentinel(seq, task_id, "stream.end") t_after_yield = time.perf_counter_ns() - logger.info( + logger.debug( "[close-debug] gateway_stream_end: reader_to_yield=%.2fms t_yielded_ns=%d task_id=%s", (t_after_yield - t_after_reader) / 1e6, t_after_yield, @@ -879,7 +879,7 @@ async def _fail(code: str, message: str) -> None: ) return False t_stub = time.perf_counter_ns() - logger.info("→ Dial-back channel ready to %s", address, extra=log_extra) + logger.debug("→ Dial-back channel ready to %s", address, extra=log_extra) def _ch_state(chan: Any) -> str: """Best-effort connectivity probe. @@ -893,7 +893,7 @@ def _ch_state(chan: Any) -> str: except Exception as exc: return f"err:{type(exc).__name__}" - logger.info( + logger.debug( "[dial-debug] channel_ready dt_init=%.3fms ch_state=%s channel_id=%s ref_count=%d cache_keys=%d", (t_stub - t_dial0) / 1e6, _ch_state(comm._channel), # noqa: SLF001 @@ -932,13 +932,13 @@ async def _outgoing() -> AsyncGenerator: nonlocal delivered_eos try: yield init_server - logger.info( + logger.debug( "→ %s sent, waiting for consumer reply before draining outputs", handshake, extra={"task_id": task_id, "mission_id": mission_id, "setup_id": setup_id}, ) await output_started.wait() - logger.info( + logger.debug( "✓ Output drain started — streaming module outputs to consumer", extra={"task_id": task_id, "mission_id": mission_id, "setup_id": setup_id}, ) @@ -998,14 +998,14 @@ async def _runner_fatal(code: str, message: str) -> None: retriable = False t_pre_stream = time.perf_counter_ns() - logger.info( + logger.debug( "[dial-debug] pre_stream dt_since_ready=%.3fms ch_state=%s", (t_pre_stream - t_stub) / 1e6, _ch_state(comm._channel), # noqa: SLF001 extra=log_extra, ) try: # noqa: PLW0717 - logger.info( + logger.debug( "→ Opening BiDi to consumer %s (sending %s)", address, handshake, @@ -1034,7 +1034,7 @@ async def _runner_fatal(code: str, message: str) -> None: except StopAsyncIteration: break except asyncio.TimeoutError: - logger.info( + logger.debug( "Consumer didn't close response stream within %.1fs after stream.end — closing BiDi", grace, extra=log_extra, @@ -1046,7 +1046,7 @@ async def _runner_fatal(code: str, message: str) -> None: if first and resume: limit = get_gateway_settings().stream.from_seq_limit resume_cursor = min(upstream.from_seq, limit) - logger.info( + logger.debug( "← Consumer resume cursor=%d received — resuming output (no re-run)", resume_cursor, extra=log_extra, @@ -1058,7 +1058,7 @@ async def _runner_fatal(code: str, message: str) -> None: if not (upstream.data and len(upstream.data.fields) > 0): continue if first: - logger.info( + logger.debug( "← First consumer reply received — starting module runner", extra=log_extra, ) diff --git a/src/digitalkin/grpc_servers/m2m_call_registry.py b/src/digitalkin/grpc_servers/m2m_call_registry.py index e3ee079b..62564004 100644 --- a/src/digitalkin/grpc_servers/m2m_call_registry.py +++ b/src/digitalkin/grpc_servers/m2m_call_registry.py @@ -188,7 +188,7 @@ async def handle_dial_back_receive( "mission_id": handle.mission_id, "target_key": handle.target_key, } - logger.info("[m2m-dialback] dial-back received, replying with query", extra=log_extra) + logger.debug("[m2m-dialback] dial-back received, replying with query", extra=log_extra) yield gateway_pb2.StreamClient(from_seq=0, task_id=task_id, data=handle.query) try: diff --git a/src/digitalkin/grpc_servers/module_servicer.py b/src/digitalkin/grpc_servers/module_servicer.py index 9fd31556..1bc2209a 100644 --- a/src/digitalkin/grpc_servers/module_servicer.py +++ b/src/digitalkin/grpc_servers/module_servicer.py @@ -214,7 +214,7 @@ async def get_or_build_tool_cache( # ``module_servicer.py:367``) clears the entry. if value is not None: self.set_tool_cache(setup_id, value) - logger.info("tool cache built for setup '%s'", setup_id) + logger.debug("tool cache built for setup '%s'", setup_id) fut.set_result(value) except Exception as exc: fut.set_exception(exc) @@ -290,14 +290,10 @@ async def _check_setup_access(self, setup_id: str) -> None: allowed = await self.user_profile.check_resource_access(user_profile_pb2.RESOURCE_TYPE_SETUP, setup_id) ids = RequestContext.current() if not allowed: - logger.info( - "[VALIDATE AC1] setup access DENIED: setup_id=%s", setup_id, extra=ids - ) # TODO(validate): remove after prod validation + logger.warning("[VALIDATE AC1] setup access DENIED: setup_id=%s", setup_id, extra=ids) msg = f"access denied to setup {setup_id}" raise PermissionDeniedError(msg) - logger.info( - "[VALIDATE AC1] setup access granted: setup_id=%s", setup_id, extra=ids - ) # TODO(validate): remove after prod validation + logger.debug("[VALIDATE AC1] setup access granted: setup_id=%s", setup_id, extra=ids) async def resolve_setup(self, setup_id: str, mission_id: str) -> SetupVersionData: """Return setup version data from cache or remote service. @@ -375,7 +371,7 @@ async def ConfigSetupModule( Raises: ServicerError: if the setup data is not returned or job creation fails. """ - logger.info( + logger.debug( "ConfigSetupVersion called for module '%s' setup_version=%s", self.module_class.__name__, request.setup_version.id, @@ -385,9 +381,7 @@ async def ConfigSetupModule( if not await self.user_profile.check_resource_access( user_profile_pb2.RESOURCE_TYPE_SETUP, setup_version.setup_id ): - logger.info( - "[VALIDATE AC1] setup config access DENIED: setup_id=%s", setup_version.setup_id - ) # TODO(validate): remove after prod validation + logger.warning("[VALIDATE AC1] setup config access DENIED: setup_id=%s", setup_version.setup_id) context.set_code(grpc.StatusCode.PERMISSION_DENIED) context.set_details(f"access denied to setup {setup_version.setup_id}") return lifecycle_pb2.ConfigSetupModuleResponse(success=False) diff --git a/src/digitalkin/mixins/agui_mixin.py b/src/digitalkin/mixins/agui_mixin.py index bec1519e..1768bb76 100644 --- a/src/digitalkin/mixins/agui_mixin.py +++ b/src/digitalkin/mixins/agui_mixin.py @@ -196,7 +196,7 @@ async def _handle_run_started( if not self._thread_id: self._thread_id = event.thread_id or str(uuid.uuid4()) - context.callbacks.logger.info( + context.callbacks.logger.debug( "[agui-mixin] RUN_STARTED thread_id=%s run_id=%s event_run_id=%s event_thread_id=%s metadata=%s", self._thread_id, self._run_id, @@ -273,7 +273,7 @@ async def _handle_run_completed( ) -> None: """Handle run completed event - emit AG-UI RunFinished.""" run_id = self._run_id or event.run_id or str(uuid.uuid4()) - context.callbacks.logger.info( + context.callbacks.logger.debug( "[agui-mixin] RUN_FINISHED thread_id=%s event_run_id=%s self._run_id=%s resolved=%s metadata=%s", self._thread_id, event.run_id, diff --git a/src/digitalkin/models/module/module_context.py b/src/digitalkin/models/module/module_context.py index eb91d988..9c14e1dc 100644 --- a/src/digitalkin/models/module/module_context.py +++ b/src/digitalkin/models/module/module_context.py @@ -28,6 +28,17 @@ from digitalkin.services.task_manager.task_manager_strategy import TaskManagerStrategy from digitalkin.services.user_profile.user_profile_strategy import UserProfileStrategy +# Lifecycle/handshake sentinels carried on ``root.protocol``: never domain output. Shared by +# ModuleToolkit._find_successful_response (skips them when picking the tool's result) and +# GrpcCommunication.call_module (excludes them from chunks_seen). +STREAM_SENTINEL_PROTOCOLS: frozenset[str] = frozenset({ + "stream.start", + "stream.init", + "stream.resume", + "stream.end", + "stream.error", +}) + class Session(SimpleNamespace): """Session data container with mandatory setup_id and mission_id.""" @@ -37,6 +48,7 @@ class Session(SimpleNamespace): setup_id: str setup_version_id: str timezone: tzinfo + cancelled: bool def __init__( self, @@ -75,6 +87,9 @@ def __init__( super().__init__(**kwargs) + # Set by `_run_lifecycle` on cancellation; toolkits refuse further calls. + self.cancelled = False + def current_ids(self) -> dict[str, str]: """Return current session ids as a dictionary. diff --git a/src/digitalkin/models/module/setup_types.py b/src/digitalkin/models/module/setup_types.py index d99acc37..835baceb 100644 --- a/src/digitalkin/models/module/setup_types.py +++ b/src/digitalkin/models/module/setup_types.py @@ -527,7 +527,7 @@ async def _collect_from_tool_ref( infos = await tool_ref.resolve(registry, communication, trim=False) for info in infos: self.resolved_tools[info.setup_id] = info - logger.info("Resolved tool '%s' -> module_id=%s", info.setup_id, info.module_id) + logger.debug("Resolved tool '%s' -> module_id=%s", info.setup_id, info.module_id) except Exception: logger.exception("Failed to resolve ToolReference '%s'", field_name) diff --git a/src/digitalkin/modules/_base_module.py b/src/digitalkin/modules/_base_module.py index b4bf20da..17854c09 100644 --- a/src/digitalkin/modules/_base_module.py +++ b/src/digitalkin/modules/_base_module.py @@ -572,6 +572,7 @@ async def _run_lifecycle( logger.info("Module %s finished", self.name, extra=self.context.session.current_ids()) except asyncio.CancelledError: self._status = ModuleStatus.CANCELLED + self.context.session.cancelled = True logger.info("Module %s cancelled", self.name, extra=self.context.session.current_ids()) raise except PermissionDeniedError as e: @@ -757,7 +758,7 @@ async def stop(self) -> None: t3 = time.perf_counter_ns() self._status = ModuleStatus.STOPPED ids = self.context.session.current_ids() - logger.info( + logger.debug( "[close-debug] module.stop: cleanup=%.2fms flush=%.2fms eos=%.2fms " "total=%.2fms t_done_ns=%d task_id=%s mission_id=%s", cleanup_ms, diff --git a/src/digitalkin/services/communication/grpc_communication.py b/src/digitalkin/services/communication/grpc_communication.py index c97c15dc..e96958c3 100644 --- a/src/digitalkin/services/communication/grpc_communication.py +++ b/src/digitalkin/services/communication/grpc_communication.py @@ -336,6 +336,11 @@ async def call_module( # noqa: C901, PLR0912, PLR0914, PLR0915 M2MTargetUnavailable: Target's breaker is open. M2MCallTimeout: Output queue stalled past ``call_timeout_s``. """ + # Local import: digitalkin.models.module.module_context pulls in this package + # (services.communication) at import time via CommunicationStrategy, so a + # module-level import here would be circular. + from digitalkin.models.module.module_context import STREAM_SENTINEL_PROTOCOLS + if self._m2m_calls is None: msg = ( "call_module needs an M2MCallRegistry wired into GrpcCommunication. " @@ -405,13 +410,13 @@ async def call_module( # noqa: C901, PLR0912, PLR0914, PLR0915 if not task_id: msg = f"backend returned no task_id from AssociateTask (parent={parent_task_id})" raise RuntimeError(msg) # noqa: TRY301 - logger.info( - "[VALIDATE AT2] AssociateTask minted: parent=%s child=%s target=%s", + logger.debug( + "[m2m] AssociateTask minted: parent=%s child=%s target=%s", parent_task_id, task_id, target_key, extra=log_extra, - ) # TODO(validate): remove after prod validation + ) log_extra["task_id"] = task_id timer.mark("associate_task") last_mark = "associate_task" @@ -469,7 +474,7 @@ async def call_module( # noqa: C901, PLR0912, PLR0914, PLR0915 breaker.record_failure() msg = f"target {target_key} rejected StartStream task_id={task_id}" raise RuntimeError(msg) - logger.info("[m2m] StartStream accepted task_id=%s", task_id, extra=log_extra) + logger.debug("[m2m] StartStream accepted task_id=%s", task_id, extra=log_extra) first_seen = False error_observed = False @@ -491,7 +496,22 @@ async def call_module( # noqa: C901, PLR0912, PLR0914, PLR0915 break now_ns = time.perf_counter_ns() - chunks_seen += 1 + + root_field = item.fields.get("root") if item.fields else None + protocol_value = "" + if root_field is not None: + proto_field = root_field.struct_value.fields.get("protocol") + protocol_value = proto_field.string_value if proto_field is not None else "" + if protocol_value == "stream.error": + fatal_field = root_field.struct_value.fields.get("fatal") + if fatal_field is not None and fatal_field.bool_value: + error_observed = True + + # Sentinel check first: the dial-back handshake frame (stream.start / + # stream.init / stream.resume) is not domain output, so it must not + # inflate chunks_seen. + if protocol_value not in STREAM_SENTINEL_PROTOCOLS: + chunks_seen += 1 depth = output_queue.qsize() max_qdepth = max(max_qdepth, depth) if not first_seen: @@ -502,14 +522,6 @@ async def call_module( # noqa: C901, PLR0912, PLR0914, PLR0915 gaps_ns.append(now_ns - last_chunk_ns) last_chunk_ns = now_ns - root_field = item.fields.get("root") if item.fields else None - if root_field is not None: - proto_field = root_field.struct_value.fields.get("protocol") - protocol_value = proto_field.string_value if proto_field is not None else "" - if protocol_value == "stream.error": - fatal_field = root_field.struct_value.fields.get("fatal") - if fatal_field is not None and fatal_field.bool_value: - error_observed = True if callback: await callback(item) yield item diff --git a/src/digitalkin/services/secret/grpc_secret.py b/src/digitalkin/services/secret/grpc_secret.py index 3409b133..ec6a7637 100644 --- a/src/digitalkin/services/secret/grpc_secret.py +++ b/src/digitalkin/services/secret/grpc_secret.py @@ -58,17 +58,17 @@ async def get_secret(self) -> dict[str, Any] | None: request = user_profile_pb2.GetSetupSecretRequest(setup_id=self.setup_id, mission_id=self.mission_id) response = await self.exec_grpc_query("GetSetupSecret", request) if not response.success: - logger.info( - "[VALIDATE SC1] secret fetch: setup_id=%s mission_id=%s success=False", + logger.warning( + "secret fetch: setup_id=%s mission_id=%s success=False", self.setup_id, self.mission_id, - ) # TODO(validate): remove after prod validation + ) return None secret = ProtoUtils.proto_to_dict(response.secret, with_defaults=True) - logger.info( - "[VALIDATE SC1] secret fetch: setup_id=%s mission_id=%s success=True keys=%d", + logger.debug( + "secret fetch: setup_id=%s mission_id=%s success=True keys=%d", self.setup_id, self.mission_id, len(secret), - ) # TODO(validate): remove after prod validation + ) return secret diff --git a/tests/community/agno/test_module_toolkit.py b/tests/community/agno/test_module_toolkit.py index 436ca99f..bcfbf508 100644 --- a/tests/community/agno/test_module_toolkit.py +++ b/tests/community/agno/test_module_toolkit.py @@ -14,6 +14,7 @@ import asyncio import json +import logging from types import SimpleNamespace from unittest.mock import AsyncMock, MagicMock @@ -25,13 +26,14 @@ from digitalkin.community.agno.module_toolkit import ModuleToolkit from digitalkin.models.module.ag_ui import AgUiOutput +from digitalkin.services.communication.exceptions import ToolCallError def _toolkit() -> ModuleToolkit: """Build a ModuleToolkit without running __init__ (which needs a live context).""" toolkit = ModuleToolkit.__new__(ModuleToolkit) - toolkit._tool_module_info = MagicMock(module_id="mod_1", setup_id="setup_1") - toolkit._context = MagicMock(session=MagicMock(job_id="job_1")) + toolkit._tool_module_info = MagicMock(module_id="mod_1", setup_id="setup_1", slug="mod_1") + toolkit._context = MagicMock(session=SimpleNamespace(job_id="job_1", cancelled=False)) return toolkit @@ -97,6 +99,76 @@ def test_frames_without_root_are_skipped(self): def test_empty_results_returns_none(self): assert ModuleToolkit._find_successful_response([]) is None + def test_stream_resume_handshake_is_skipped(self): + """`stream.resume` is a handshake sentinel, like `stream.init` — not a domain frame.""" + results = [ + {"root": {"protocol": "search", "results": [1]}}, + {"root": {"protocol": "stream.resume"}}, + {"root": {"protocol": "stream.end"}}, + ] + resp = ModuleToolkit._find_successful_response(results) + assert resp == {"root": {"protocol": "search", "results": [1]}} + + +class TestFatalStreamError: + """A fatal `stream.error` anywhere in the stream means the tool did not finish. + + Progress lines (e.g. tool-rag-methods `add_documents` streaming + "Indexing i/n: name" before each file) or a gateway idle-timeout error must + not surface the last progress line, or any later stray frame, as a success. + A non-fatal `stream.error` (the target module kept the stream open, per + `M2MCallRegistry.handle_dial_back_receive`) does not fail the call on its own. + """ + + def test_progress_then_fatal_error_is_a_failure(self) -> None: + """A progress line followed by stream.error must not be returned as the tool result.""" + results = [ + {"root": {"protocol": "tool_content", "content": "Indexing 7/10: g.pdf"}}, + {"root": {"protocol": "stream.error", "code": "INTERNAL", "message": "boom", "fatal": True}}, + {"root": {"protocol": "stream.end"}}, + ] + assert ModuleToolkit._find_successful_response(results) is None + + def test_progress_then_receipt_is_the_receipt(self) -> None: + results = [ + {"root": {"protocol": "tool_content", "content": "Indexing 1/1: a.pdf"}}, + {"root": {"protocol": "tool_content", "content": "receipt"}}, + {"root": {"protocol": "stream.end"}}, + ] + assert ModuleToolkit._find_successful_response(results)["root"]["content"] == "receipt" + + def test_non_fatal_error_then_receipt_is_the_receipt(self) -> None: + """A non-fatal stream.error (the module kept the stream open) does not fail the call.""" + results = [ + {"root": {"protocol": "stream.error", "code": "TRANSIENT", "message": "retrying", "fatal": False}}, + {"root": {"protocol": "tool_content", "content": "receipt"}}, + {"root": {"protocol": "stream.end"}}, + ] + assert ModuleToolkit._find_successful_response(results)["root"]["content"] == "receipt" + + def test_error_before_a_stray_domain_frame_is_still_a_failure(self) -> None: + """A stream.error anywhere in the results fails the call, even if a domain + frame streamed after it (e.g. a late, out-of-order write).""" + results = [ + {"root": {"protocol": "stream.error", "code": "INTERNAL", "message": "boom", "fatal": True}}, + {"root": {"protocol": "tool_content", "content": "stray frame"}}, + ] + assert ModuleToolkit._find_successful_response(results) is None + + def test_failure_text_carries_the_stream_error_code_and_message(self) -> None: + """The JSON body handed back to the model names the failure, not a generic 'no response'.""" + toolkit = _toolkit() + results = [ + {"root": {"protocol": "tool_content", "content": "Indexing 7/10: g.pdf"}}, + {"root": {"protocol": "stream.error", "code": "INTERNAL", "message": "boom", "fatal": True}}, + {"root": {"protocol": "stream.end"}}, + ] + assert ModuleToolkit._find_successful_response(results) is None + error_msg = ModuleToolkit._extract_error_message(results) + body = json.loads(toolkit._handle_failure("add_documents", error_msg, 1.0, {})) + assert "INTERNAL" in body["error"] + assert "boom" in body["error"] + class TestExtractErrorMessage: def test_stream_error_surfaces_code_and_message(self): @@ -271,3 +343,96 @@ def test_malformed_message_is_a_no_op(self): asyncio.run(ModuleToolkit._relay_custom_event(_context(send), message)) send.assert_not_awaited() + + +class TestToolkitBuildLogging: + def test_build_emits_one_debug_line_and_no_info(self, caplog: pytest.LogCaptureFixture) -> None: + """A toolkit build is trace, not lifecycle: one DEBUG summary, nothing at INFO.""" + tool_def = SimpleNamespace( + name="read_json", + description="Read a record", + parameters_schema={"type": "object", "properties": {}}, + parameter_count=0, + parameter_names=[], + ) + + async def fn(**kwargs: object) -> None: + yield {} + + context = MagicMock() + context.create_tool_functions.return_value = [(tool_def, fn)] + info = MagicMock(setup_id="setup_1", slug="storage", module_name="Storage", tool_name="", tools=[tool_def]) + + with caplog.at_level(logging.DEBUG, logger="digitalkin"): + ModuleToolkit(context=context, tool_module_info=info) + + build_records = [r for r in caplog.records if "toolkit" in r.getMessage().lower()] + assert [r.levelno for r in build_records] == [logging.DEBUG] + assert "storage" in build_records[0].getMessage() + assert "tools=1/1" in build_records[0].getMessage() + + +class TestToolCallLogging: + def test_success_logs_once_at_info_with_argument_keys(self, caplog: pytest.LogCaptureFixture) -> None: + """The completion line is the only INFO record and carries the sorted argument keys.""" + toolkit = _toolkit() + with caplog.at_level(logging.DEBUG, logger="digitalkin"): + toolkit._handle_success( + "read_json", + {"root": {"protocol": "tool_content", "content": "x"}}, + 12.5, + {"collection": "a", "record_id": "b"}, + ) + info = [r for r in caplog.records if r.levelno == logging.INFO] + assert len(info) == 1 + assert "read_json" in info[0].getMessage() + assert "args=['collection', 'record_id']" in info[0].getMessage() + + +class TestCancelledTask: + @pytest.mark.asyncio + async def test_wrapper_refuses_after_cancel(self, caplog: pytest.LogCaptureFixture) -> None: + toolkit = _toolkit() + toolkit._context.session.cancelled = True + toolkit._timeout = 1.0 + called = False + + async def fn(**kwargs: object): + nonlocal called + called = True + yield {} + + tool_def = SimpleNamespace( + name="write_json", description="", parameters_schema={}, parameter_count=0, parameter_names=[] + ) + wrapper = toolkit._create_tool_wrapper(tool_def, fn) + with caplog.at_level(logging.WARNING, logger="digitalkin"), pytest.raises(asyncio.CancelledError): + await wrapper() + assert called is False + assert any("refused after task cancellation" in r.getMessage() for r in caplog.records) + + +class TestWrapperFatalStreamError: + """The real fatal path: `_create_single_tool_function` raises `ToolCallError`, not a + `stream.error` dict — the wrapper must still surface it as a failure with code+message.""" + + @pytest.mark.asyncio + async def test_tool_call_error_becomes_a_failure_with_code_and_message(self) -> None: + toolkit = _toolkit() + toolkit._timeout = 1.0 + + async def fn(**kwargs: object): + msg = "[SETUP_ACCESS_DENIED] denied" + raise ToolCallError(msg) + yield {} # pragma: no cover # unreachable; makes this an async generator + + tool_def = SimpleNamespace( + name="search", description="", parameters_schema={}, parameter_count=0, parameter_names=[] + ) + wrapper = toolkit._create_tool_wrapper(tool_def, fn) + result = await wrapper() + + assert isinstance(result, str) + body = json.loads(result) + assert "SETUP_ACCESS_DENIED" in body["error"] + assert "denied" in body["error"] diff --git a/tests/core/test_task_executor.py b/tests/core/test_task_executor.py index b99fda8c..efae99f8 100644 --- a/tests/core/test_task_executor.py +++ b/tests/core/test_task_executor.py @@ -8,6 +8,7 @@ import asyncio import contextlib +import logging import time from typing import NoReturn from unittest.mock import AsyncMock, Mock @@ -77,6 +78,7 @@ async def test_main_task_completes_successfully( self, task_executor: TaskExecutor, mock_base_module: Mock, + caplog: pytest.LogCaptureFixture, ) -> None: """Test executor when main task completes successfully.""" task_id = "main_success" @@ -90,11 +92,12 @@ async def main_coro() -> None: await asyncio.sleep(0.1) execution_log.append("main_end") - supervisor = await task_executor.execute_task( - task_id, mission_id, main_coro(), session - ) + with caplog.at_level(logging.INFO, logger="digitalkin"): + supervisor = await task_executor.execute_task( + task_id, mission_id, main_coro(), session + ) - await supervisor + await supervisor assert session.status == "completed" assert "main_start" in execution_log @@ -102,6 +105,13 @@ async def main_coro() -> None: assert session.started_at is not None assert session.completed_at is not None + # The whole run must produce exactly the two lifecycle lines at INFO: + # "Task completed" (status transition) and "Task done" (final summary). + info = [r.getMessage() for r in caplog.records if r.levelno == logging.INFO and r.name.startswith("digitalkin")] + lifecycle = [m for m in info if m.startswith("Task completed") or m.startswith("Task done")] + assert len(lifecycle) == 2, info + assert info == lifecycle, info + @pytest.mark.asyncio async def test_main_task_completion_timing_accuracy( self, diff --git a/tests/gateway/test_dial_consumer_full_duplex.py b/tests/gateway/test_dial_consumer_full_duplex.py index 0d60e1d7..5f26aa81 100644 --- a/tests/gateway/test_dial_consumer_full_duplex.py +++ b/tests/gateway/test_dial_consumer_full_duplex.py @@ -16,6 +16,7 @@ from __future__ import annotations import asyncio +import logging from typing import Any import grpc.aio @@ -108,8 +109,16 @@ async def test_unbounded_upstream_inputs(self, gateway_with_runner) -> None: finally: await server.stop(grace=0.1) - async def test_unbounded_outputs(self, gateway_with_runner) -> None: - """100 outputs pumped through task:{id}:stream all reach the consumer.""" + async def test_unbounded_outputs(self, gateway_with_runner, caplog: pytest.LogCaptureFixture) -> None: + """100 outputs pumped through task:{id}:stream all reach the consumer. + + Also covers the full StartStream -> dial-back -> drain cycle's log + volume: this GatewayServicer path never reaches ``TaskExecutor`` + (the module runner is faked), so the only INFO line from + ``digitalkin.*`` loggers for the whole cycle is the "Task accepted" + lifecycle line — everything else (dial-back scheduling, channel + readiness, handshake, drain start, cursor timing) is DEBUG-level trace. + """ gateway, redis = gateway_with_runner n_outputs = 100 servicer = _FakeConsumerServicer(query_data={"q": "go"}) @@ -129,14 +138,15 @@ async def test_unbounded_outputs(self, gateway_with_runner) -> None: await redis.xadd(stream_key, {"pb": s.SerializeToString(), "seq": str(i + 1)}) await redis.xadd(stream_key, {"eos": b"true"}) - ctx = _mock_context({"x-client-address": f"127.0.0.1:{port}"}) - await gateway.StartStream(_start_request(task_id), ctx) + with caplog.at_level(logging.INFO, logger="digitalkin"): + ctx = _mock_context({"x-client-address": f"127.0.0.1:{port}"}) + await gateway.StartStream(_start_request(task_id), ctx) - # Wait until the consumer sees stream.end on the wire. - for _ in range(200): - if any(_protocol_of(m) == "stream.end" for m in servicer.received): - break - await asyncio.sleep(0.1) + # Wait until the consumer sees stream.end on the wire. + for _ in range(200): + if any(_protocol_of(m) == "stream.end" for m in servicer.received): + break + await asyncio.sleep(0.1) ticks = [ int(m.data.fields["root"].struct_value.fields["i"].number_value) @@ -146,5 +156,16 @@ async def test_unbounded_outputs(self, gateway_with_runner) -> None: assert ticks == list(range(n_outputs)) protos = [_protocol_of(m) for m in servicer.received] assert protos[-1] == "stream.end" + + # This cycle never reaches TaskExecutor (fake module runner), so + # "Task accepted" is the only lifecycle line — and the only INFO line. + info = [ + r.getMessage() + for r in caplog.records + if r.levelno == logging.INFO and r.name.startswith("digitalkin") + ] + lifecycle = [m for m in info if m.startswith("Task accepted")] + assert len(lifecycle) == 1, info + assert info == lifecycle, info finally: await server.stop(grace=0.1) diff --git a/tests/gateway/test_m2m_call_module.py b/tests/gateway/test_m2m_call_module.py index e10f1d31..f820f65e 100644 --- a/tests/gateway/test_m2m_call_module.py +++ b/tests/gateway/test_m2m_call_module.py @@ -13,6 +13,7 @@ from __future__ import annotations import asyncio +import logging from collections.abc import AsyncIterator from typing import Any from unittest.mock import AsyncMock, MagicMock @@ -191,6 +192,7 @@ async def test_round_trip_mints_via_backend_then_streams( backend_server: tuple[_FakeBackendGateway, str, int], callee_server: tuple[_FakeCalleeGatewayServicer, str, int], caller_gateway: tuple[GatewayServicer, str, int], + caplog: pytest.LogCaptureFixture, ) -> None: backend, backend_host, backend_port = backend_server callee_servicer, callee_host, callee_port = callee_server @@ -208,18 +210,23 @@ async def test_round_trip_mints_via_backend_then_streams( outputs: list[Any] = [] token = RequestContext.bind(task_id="task:parent") try: - async for out_struct in comm.call_module( - module_address=callee_host, - module_port=callee_port, - input_data={"root": {"protocol": "transform", "text": "hello"}}, - setup_id="setups:test", - mission_id="missions:test", - ): - outputs.append(out_struct) + with caplog.at_level(logging.DEBUG, logger="digitalkin"): + async for out_struct in comm.call_module( + module_address=callee_host, + module_port=callee_port, + input_data={"root": {"protocol": "transform", "text": "hello"}}, + setup_id="setups:test", + mission_id="missions:test", + ): + outputs.append(out_struct) finally: RequestContext.reset(token) await comm.close() + # The m2m handshake lines describe the wire protocol, not an outcome — DEBUG only. + info_messages = [r.getMessage() for r in caplog.records if r.levelno == logging.INFO] + assert not any("AssociateTask minted" in m or "StartStream accepted" in m for m in info_messages) + domain = [o for o in outputs if o.fields["root"].struct_value.fields["protocol"].string_value == "transform"] assert [o.fields["root"].struct_value.fields["value"].string_value for o in domain] == [ "hello-1", diff --git a/tests/gateway/test_m2m_end_to_end.py b/tests/gateway/test_m2m_end_to_end.py index ee51e0a8..932ee80f 100644 --- a/tests/gateway/test_m2m_end_to_end.py +++ b/tests/gateway/test_m2m_end_to_end.py @@ -12,8 +12,11 @@ register is rejected ``UNAUTHENTICATED "Invalid or inactive task"`` — the regression test reproduces the prod bug that motivated the backend mint. -Assertions are tied to the prod validation markers: ``[VALIDATE AT2]`` (caller -mint) and ``[VALIDATE AC1]`` (setup access verdict). +Assertions are tied to the ``[VALIDATE AC1]`` prod validation marker (setup +access verdict). The caller mint is asserted directly on the backend's fake +``AssociateTask`` state (``state.mint_count`` etc.) rather than a log marker: +the handshake line that used to carry ``[VALIDATE AT2]`` is DEBUG-only now +that prod validation is complete. """ from __future__ import annotations @@ -130,19 +133,28 @@ def _clear_singletons() -> Generator[None]: @pytest.fixture def digitalkin_records() -> Generator[list[logging.LogRecord]]: - """Capture 'digitalkin' logger records (the [VALIDATE ...] markers) at INFO. + """Capture 'digitalkin' logger records (the [VALIDATE ...] markers) at DEBUG. + + The "granted" marker is DEBUG-only trace now that prod validation is complete + (only the "DENIED" marker stays a WARNING security signal), so the logger's + own effective level is lowered to DEBUG for the duration of the test — + otherwise the root logger's INFO level (set in ``tests/conftest.py``) would + filter it out before it ever reaches this handler. Yields: The captured records list, live-updated while the test runs. """ records: list[logging.LogRecord] = [] handler = logging.Handler() - handler.setLevel(logging.INFO) + handler.setLevel(logging.DEBUG) handler.emit = records.append # type: ignore[method-assign] lg = logging.getLogger("digitalkin") + previous_level = lg.level + lg.setLevel(logging.DEBUG) lg.addHandler(handler) yield records lg.removeHandler(handler) + lg.setLevel(previous_level) def _marker_lines(records: list[logging.LogRecord], marker: str) -> list[str]: @@ -452,11 +464,9 @@ async def test_full_tool_call_child_authenticated_and_output_streamed( assert "healthcheck_ping" in protocols, [json_format.MessageToDict(o) for o in outputs] assert _stream_errors(outputs) == [] - # 4. The prod validation markers traced the whole chain. - at2 = _marker_lines(digitalkin_records, "[VALIDATE AT2]") - assert len(at2) == 1 - assert f"parent={PARENT_TASK_ID}" in at2[0] - assert "child=child-1" in at2[0] + # 4. The prod validation marker traced the access verdict (the mint side of + # the chain is asserted above via state.mint_* — its own log line is + # DEBUG-only now that prod validation is complete, so it carries no marker). ac1 = _marker_lines(digitalkin_records, "[VALIDATE AC1]") assert any("setup access granted" in line and SETUP_ID in line for line in ac1) diff --git a/tests/gateway/test_m2m_resilience.py b/tests/gateway/test_m2m_resilience.py index 8be43f92..70504ce4 100644 --- a/tests/gateway/test_m2m_resilience.py +++ b/tests/gateway/test_m2m_resilience.py @@ -13,6 +13,7 @@ from __future__ import annotations import asyncio +import logging import time from collections.abc import AsyncIterator from typing import Any @@ -289,6 +290,88 @@ async def test_output_queue_silence_raises(self, monkeypatch: pytest.MonkeyPatch ): pass + async def test_timeout_after_handshake_only_reports_chunks_seen_zero( + self, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture + ) -> None: + """Only the dial-back handshake frame arrived; it must not inflate ``chunks_seen``.""" + monkeypatch.setenv("DIGITALKIN_M2M_CALL_TIMEOUT_S", "0.15") + get_gateway_settings.cache_clear() + gw = _gw() + comm = _comm(gw) + + stub_mock = MagicMock() + stub_mock.StartStream = AsyncMock( + return_value=gateway_pb2.StartStreamResponse(accepted=True, task_id="tid"), + ) + stub_mock.SendSignal = AsyncMock() + comm._get_or_create_channel = MagicMock(return_value=MagicMock()) # type: ignore[method-assign] + comm._get_or_create_stub = MagicMock(return_value=stub_mock) # type: ignore[method-assign] + + async def _drive() -> None: + async for _ in comm.call_module( + module_address="127.0.0.1", + module_port=9999, + input_data={"root": {"protocol": "x"}}, + setup_id="setups:test", + mission_id="missions:test", + ): + pass + + task = asyncio.create_task(_drive()) + await asyncio.sleep(0.02) # let call_module register and call StartStream + entry = gw._m2m.get("tid") + assert entry is not None + # Only the dial-back handshake frame arrives; no domain output ever does. + entry.output_queue.put_nowait(_struct({"root": {"protocol": "stream.init"}})) + + with caplog.at_level(logging.WARNING, logger="digitalkin"), pytest.raises(M2MCallTimeout): + await task + + records = [r for r in caplog.records if "call_module_failed" in r.getMessage()] + assert len(records) == 1 + assert "chunks_seen=0" in records[0].getMessage() + + async def test_timeout_after_resume_handshake_only_reports_chunks_seen_zero( + self, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture + ) -> None: + """`stream.resume` is a handshake sentinel too; it must not inflate ``chunks_seen``.""" + monkeypatch.setenv("DIGITALKIN_M2M_CALL_TIMEOUT_S", "0.15") + get_gateway_settings.cache_clear() + gw = _gw() + comm = _comm(gw) + + stub_mock = MagicMock() + stub_mock.StartStream = AsyncMock( + return_value=gateway_pb2.StartStreamResponse(accepted=True, task_id="tid"), + ) + stub_mock.SendSignal = AsyncMock() + comm._get_or_create_channel = MagicMock(return_value=MagicMock()) # type: ignore[method-assign] + comm._get_or_create_stub = MagicMock(return_value=stub_mock) # type: ignore[method-assign] + + async def _drive() -> None: + async for _ in comm.call_module( + module_address="127.0.0.1", + module_port=9999, + input_data={"root": {"protocol": "x"}}, + setup_id="setups:test", + mission_id="missions:test", + ): + pass + + task = asyncio.create_task(_drive()) + await asyncio.sleep(0.02) # let call_module register and call StartStream + entry = gw._m2m.get("tid") + assert entry is not None + # Only the resume handshake frame arrives; no domain output ever does. + entry.output_queue.put_nowait(_struct({"root": {"protocol": "stream.resume"}})) + + with caplog.at_level(logging.WARNING, logger="digitalkin"), pytest.raises(M2MCallTimeout): + await task + + records = [r for r in caplog.records if "call_module_failed" in r.getMessage()] + assert len(records) == 1 + assert "chunks_seen=0" in records[0].getMessage() + class TestCancellation: """Cancelled ``call_module`` sends a best-effort ``SendSignal(CANCEL)``.""" diff --git a/tests/gateway/test_stream_error_propagation.py b/tests/gateway/test_stream_error_propagation.py index aa6bed20..fa270278 100644 --- a/tests/gateway/test_stream_error_propagation.py +++ b/tests/gateway/test_stream_error_propagation.py @@ -276,6 +276,19 @@ class _Strict(BaseModel): raise AssertionError("model_validate unexpectedly succeeded") +def _real_type_validation_error() -> ValidationError: + """Produce a genuine pydantic ValidationError from a present field with the wrong type.""" + + class _Strict(BaseModel): + patch: dict[str, Any] + + try: + _Strict.model_validate({"patch": "not-a-dict"}) + except ValidationError as exc: + return exc + raise AssertionError("model_validate unexpectedly succeeded") + + @SKIP_NO_FAKEREDIS class TestValidationErrorPhases: @staticmethod @@ -354,6 +367,39 @@ async def _on_fatal(code: str, message: str) -> None: finally: await redis.close() + async def test_input_validation_error_reports_field_and_reason(self) -> None: + """A type error on a present field must name the field and reason, not just top-level keys.""" + from digitalkin.core.task_manager.module_runner import ModuleRunner + + redis = _FakeRedisClient() + try: + servicer = self._servicer() + servicer.module_class.create_input_model = MagicMock(side_effect=_real_type_validation_error()) + servicer.module_class._extended_input_format = None + servicer.module_class.input_format = type("FakeInput", (), {}) + runner = ModuleRunner(redis_client=redis, servicer=servicer) # type: ignore[arg-type] + + received: list[tuple[str, str]] = [] + + async def _on_fatal(code: str, message: str) -> None: + received.append((code, message)) + + await runner.run( + struct_pb2.Struct(), + task_id="task_val", + setup_id="setups:s1", + mission_id="missions:m1", + on_fatal=_on_fatal, + ) + + assert len(received) == 1 + code, message = received[0] + assert code == StreamErrorCode.INPUT_VALIDATION_ERROR.value + assert "patch" in message + assert "dict" in message + finally: + await redis.close() + # =========================================================================== # GrpcCommunication.stream_error helper diff --git a/tests/modules/test_base_module_lifecycle.py b/tests/modules/test_base_module_lifecycle.py index 570a0b2e..16ba27de 100644 --- a/tests/modules/test_base_module_lifecycle.py +++ b/tests/modules/test_base_module_lifecycle.py @@ -357,6 +357,7 @@ async def test_cancel_sets_cancelled(self) -> None: await module._run_lifecycle(_LcInputModel(root=_LcInputTrigger()), _LcSetupModel()) assert module.status == ModuleStatus.CANCELLED + assert module.context.session.cancelled is True @pytest.mark.unit @pytest.mark.regression diff --git a/uv.lock b/uv.lock index b978dacd..dc87af9f 100644 --- a/uv.lock +++ b/uv.lock @@ -715,7 +715,7 @@ wheels = [ [[package]] name = "digitalkin" -version = "1.0.3" +version = "1.0.4.dev0" source = { editable = "." } dependencies = [ { name = "ag-ui-protocol" }, @@ -909,7 +909,7 @@ name = "exceptiongroup" version = "1.3.1" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "typing-extensions" }, + { name = "typing-extensions", marker = "python_full_version < '3.12'" }, ] sdist = { url = "https://files.pythonhosted.org/packages/50/79/66800aadf48771f6b62f7eb014e352e5d06856655206165d775e675a02c9/exceptiongroup-1.3.1.tar.gz", hash = "sha256:8b412432c6055b0b7d14c310000ae93352ed6754f70fa8f7c34141f91c4e3219", size = 30371, upload-time = "2025-11-21T23:01:54.787Z" } wheels = [ @@ -1551,7 +1551,7 @@ name = "importlib-metadata" version = "9.0.0" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "zipp" }, + { name = "zipp", marker = "python_full_version < '3.12'" }, ] sdist = { url = "https://files.pythonhosted.org/packages/a9/01/15bb152d77b21318514a96f43af312635eb2500c96b55398d020c93d86ea/importlib_metadata-9.0.0.tar.gz", hash = "sha256:a4f57ab599e6a2e3016d7595cfd72eb4661a5106e787a95bcc90c7105b831efc", size = 56405, upload-time = "2026-03-20T06:42:56.999Z" } wheels = [