Fix(logging): cut per-task log noise and close the incident gaps from the 2026-09-03 trace

.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)
     (?P<major>0|[1-9]\\d*)\\.
     (?P<minor>0|[1-9]\\d*)\\.

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" ]

src/digitalkin/__version__.py

@@ -5,4 +5,4 @@
 try:
     __version__ = version("digitalkin")
 except PackageNotFoundError:
-    __version__ = "1.0.3"
+    __version__ = "1.0.4.dev0"

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

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()),

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]]:

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]:

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),

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)