fix(subagents): stamp UTC-aware datetimes on SubagentResult lifecycle (#5153)

## Why

DeerFlow declares one timestamp convention in deerflow/utils/time.py: every lifecycle timestamp is UTC (now_iso / datetime.now(UTC)). SubagentResult writers in subagents/executor.py still used naive datetime.now(), so on any non-UTC host the in-memory lifecycle metadata (started_at / completed_at) was local wall-clock time. The sibling durable-batch path (subagents/batch_service.py) already stamps datetime.now(UTC), so the same run model carried two different conventions depending on which path wrote it.

## What changed

- Added an executor-local _utcnow() helper that stamps datetime.now(UTC).

- SubagentResult.completed_at default in try_set_terminal(), result.started_at in _aexecute(), and the started_at default in _aexecute_admitted() now route through _utcnow().

- Explicit caller-supplied timestamps (completed_at=...) still pass through unchanged.

- Added regression tests asserting the default writers produce UTC-aware datetimes.

## Surface area

- [x] Backend runtime (deerflow.subagents.executor) - internal dataclass lifecycle metadata, no wire format change

- [ ] Frontend UI / Backend API / Sandbox / Skills / Dependencies / Default behavior change

## Bug fix verification

- New tests: tests/test_subagent_executor.py::test_timestamp_writers_stamp_utc_aware_datetimes and test_utcnow_helper_returns_utc_aware_datetime encode the convention.

- Updated BlockingDateTime.now() in the terminal-publication-order test to mirror datetime.now's optional tz argument.

## Validation

- cd backend && python -m pytest tests/test_subagent_executor.py: 136 passed; 2 pre-existing TestBashExecutionHarvest failures reproduce identically on clean main (Windows sandbox env), unrelated to this change.

- ruff format + ruff check clean on both changed files.

## AI assistance

**Tool(s) used:** Codex (coding agent)

**How you used it:** analysis of the timestamp conventions, implementation, and regression tests authored with AI assistance; change reviewed before commit.

- [ ] I've read and understand every line of this change and take responsibility for it — it's not unreviewed AI output.

backend/packages/harness/deerflow/subagents/executor.py

@@ -13,7 +13,7 @@ from concurrent.futures import Future
 from concurrent.futures import TimeoutError as FuturesTimeoutError
 from contextvars import Context, copy_context
 from dataclasses import dataclass, field
-from datetime import datetime
+from datetime import UTC, datetime
 from enum import Enum
 from typing import TYPE_CHECKING, Any
 
@@ -67,6 +67,12 @@ _SANDBOX_LEASE_OWNER_CONTEXT_KEY = "sandbox_lease_owner_id"
 _SANDBOX_COMMAND_SCOPE_CONTEXT_KEY = "sandbox_command_scope_id"
 
 
+def _utcnow() -> datetime:
+    # SubagentResult timestamp writers must stamp UTC-aware datetimes so
+    # lifecycle metadata never depends on the host wall clock (see deerflow.utils.time).
+    return datetime.now(UTC)
+
+
 _previous_shutdown_isolated_subagent_loop = globals().get("_shutdown_isolated_subagent_loop")
 if callable(_previous_shutdown_isolated_subagent_loop):
     atexit.unregister(_previous_shutdown_isolated_subagent_loop)
@@ -242,7 +248,7 @@ class SubagentResult:
             if tool_receipts is not None:
                 self.tool_receipts = [dict(receipt) for receipt in tool_receipts]
             self.admission_failure = admission_failure
-            self.completed_at = completed_at or datetime.now()
+            self.completed_at = completed_at or _utcnow()
             self.status = status
             return True
 
@@ -1281,7 +1287,7 @@ class SubagentExecutor:
                     with result._state_lock:
                         if not result.status.is_terminal:
                             result.status = SubagentStatus.RUNNING
-                            result.started_at = datetime.now()
+                            result.started_at = _utcnow()
                     return await self._aexecute_admitted(task, result)
             except SubagentCapacityError as exc:
                 result.try_set_terminal(
@@ -1311,7 +1317,7 @@ class SubagentExecutor:
                 task_id=task_id,
                 trace_id=self.trace_id,
                 status=SubagentStatus.RUNNING,
-                started_at=datetime.now(),
+                started_at=_utcnow(),
             )
         sandbox_lease_owner_id = f"subagent:{result.task_id}"
         execution_context: dict[str, Any] | None = None

backend/tests/test_subagent_executor.py

@@ -2381,7 +2381,9 @@ class TestThreadSafety:
 
         class BlockingDateTime:
             @staticmethod
-            def now():
+            def now(tz=None):
+                # Signature mirrors datetime.now's optional tz argument: the
+                # production writer stamps UTC via datetime.now(UTC).
                 now_entered.set()
                 release_now.wait(timeout=5)
                 return completed_at
@@ -4876,3 +4878,27 @@ class TestBashExecutionHarvest:
 
         assert result.status == SubagentStatus.COMPLETED
         assert result.bash_executions is None
+
+
+def test_timestamp_writers_stamp_utc_aware_datetimes(classes):
+    """Terminal transitions must stamp UTC-aware datetimes, not naive local wall-clock values."""
+    SubagentResult = classes["SubagentResult"]
+    SubagentStatus = classes["SubagentStatus"]
+
+    result = SubagentResult(task_id="tz-check", trace_id="trace-1", status=SubagentStatus.PENDING)
+    assert result.try_set_terminal(SubagentStatus.COMPLETED, result="done")
+
+    assert result.completed_at is not None
+    assert result.completed_at.tzinfo is not None
+    assert result.completed_at.utcoffset() is not None
+    assert result.completed_at.utcoffset().total_seconds() == 0.0
+
+
+def test_utcnow_helper_returns_utc_aware_datetime(classes):
+    """The shared timestamp writer must never depend on the host wall clock."""
+    executor_module = sys.modules["deerflow.subagents.executor"]
+
+    now = executor_module._utcnow()
+    assert now.tzinfo is not None
+    assert now.utcoffset() is not None
+    assert now.utcoffset().total_seconds() == 0.0