fix(tasks): keep scheduled-task prompt cache stable

Move scheduled-task current-time context out of the system prompt and into a user-role context message so the system prompt remains stable for prompt caching. Preserve time grounding on both the agent-loop path and fallback direct-call path, with focused regression coverage.

src/task_scheduler.py

@@ -1450,19 +1450,18 @@ class TaskScheduler:
                     system_prompt = f"{char_prompt}\n\n{system_prompt}"
             except Exception:
                 pass
-        # Inject current time so the model knows what's past vs upcoming
+        # Provide current date/time as a user-role message so the system prompt
+        # stays byte-identical across runs and doesn't bust the Anthropic prompt
+        # cache on every scheduled tick (see issue #2927 and the identical fix on
+        # the interactive-chat path in src/agent_loop.py).  The message is built
+        # once here and shared by both execution paths below (agent loop and the
+        # direct fallback) so time grounding is never lost on either path.
         tz_name = _resolve_task_timezone(db, task)
         try:
-            if tz_name:
-                from zoneinfo import ZoneInfo
-                from datetime import timezone
-                now_local = _utcnow().replace(tzinfo=timezone.utc).astimezone(ZoneInfo(tz_name))
-                time_str = now_local.strftime("%A, %B %d %Y, %H:%M %Z")
-            else:
-                time_str = _utcnow().strftime("%A, %B %d %Y, %H:%M UTC")
+            from src.user_time import current_datetime_context_message_for_tz
+            _dt_msg: dict | None = current_datetime_context_message_for_tz(tz_name)
         except Exception:
-            time_str = _utcnow().strftime("%A, %B %d %Y, %H:%M UTC")
-        system_prompt = f"Current time: {time_str}\n\n{system_prompt}"
+            _dt_msg = None
 
         # Compute the disabled-tools set: the crew's enabled_tools allowlist
         # (inverted) plus the operator's global disabled_tools setting. The
@@ -1510,14 +1509,15 @@ class TaskScheduler:
                 endpoint_url, model, task, session_id,
                 system_prompt=system_prompt, disabled_tools=disabled_tools or None,
                 relevant_tools=relevant_tools,
+                datetime_context_msg=_dt_msg,
             )
         except Exception as e:
             logger.warning(f"Agent loop failed for task '{task.name}', falling back to simple call: {e}")
             from src.task_endpoint import task_llm_call_async
-            messages = [
-                {"role": "system", "content": system_prompt},
-                {"role": "user", "content": task.prompt},
-            ]
+            messages: list = [{"role": "system", "content": system_prompt}]
+            if _dt_msg:
+                messages.append(_dt_msg)
+            messages.append({"role": "user", "content": task.prompt})
             result = await task_llm_call_async(
                 messages,
                 fallback_url=endpoint_url,
@@ -1715,16 +1715,20 @@ class TaskScheduler:
                               system_prompt: str | None = None,
                               disabled_tools: set | None = None,
                               relevant_tools: set | None = None,
-                              override_user_message: str | None = None) -> str:
+                              override_user_message: str | None = None,
+                              datetime_context_msg: dict | None = None) -> str:
         """Run the full agent loop with tool access, collecting the final text."""
         from src.agent_loop import stream_agent_loop
 
         system_content = system_prompt or "You are a helpful assistant executing a scheduled task. Use available tools to complete the task thoroughly."
         user_content = override_user_message or task.prompt
-        messages = [
-            {"role": "system", "content": system_content},
-            {"role": "user", "content": user_content},
-        ]
+        # Build the message list. The datetime context message (user-role) is
+        # inserted immediately before the task prompt so the system prefix stays
+        # byte-identical and cacheable across runs (see issue #2927).
+        messages: list = [{"role": "system", "content": system_content}]
+        if datetime_context_msg:
+            messages.append(datetime_context_msg)
+        messages.append({"role": "user", "content": user_content})
 
         # Resolve headers from the endpoint's API key
         headers = {}

src/user_time.py

@@ -138,6 +138,69 @@ def current_datetime_prompt(now_utc: Optional[datetime] = None) -> str:
     )
 
 
+def current_datetime_context_message_for_tz(
+    iana_tz_name: Optional[str],
+    now_utc: Optional[datetime] = None,
+) -> Dict[str, str]:
+    """Build the current-date/time context as a user-role message, resolved
+    against an explicit IANA timezone name rather than browser ContextVars.
+
+    Unlike ``current_datetime_context_message()``, this function does not read
+    or write any ContextVar and leaves no per-request state behind — it is safe
+    to call from background tasks that have no browser request context.
+
+    Timezone resolution:
+    * ``iana_tz_name`` is a valid IANA name (e.g. ``"Europe/Berlin"``) → uses that zone.
+    * ``iana_tz_name`` is ``None`` OR resolves to an invalid zone → falls back to UTC.
+      This matches the existing scheduler behaviour: tasks without a linked crew
+      timezone render in UTC, not server-local time.
+    """
+    if now_utc is None:
+        utc_now = datetime.now(timezone.utc)
+    elif now_utc.tzinfo is None:
+        utc_now = now_utc.replace(tzinfo=timezone.utc)
+    else:
+        utc_now = now_utc.astimezone(timezone.utc)
+
+    # Resolve the display timezone — UTC fallback on any failure.
+    tz = timezone.utc
+    resolved_name: Optional[str] = None
+    if iana_tz_name:
+        try:
+            from zoneinfo import ZoneInfo
+            tz = ZoneInfo(iana_tz_name)
+            resolved_name = iana_tz_name
+        except Exception:
+            tz = timezone.utc  # invalid zone → UTC, no ContextVar touched
+
+    local_now = utc_now.astimezone(tz)
+    tomorrow = local_now + timedelta(days=1)
+
+    _utc_offset = local_now.utcoffset()
+    offset_min = int(_utc_offset.total_seconds() // 60) if _utc_offset is not None else 0
+    offset_label = f"UTC{format_utc_offset(offset_min)}"
+    tz_label = f"{resolved_name}, {offset_label}" if resolved_name else offset_label
+
+    prompt = (
+        "## Current date and time\n"
+        f"Today is {_date_label(local_now)} ({local_now.strftime('%Y-%m-%d')}). "
+        f"Local time is {_clock_label(local_now)} ({tz_label}); "
+        f"current UTC time is {utc_now.strftime('%H:%M')}.\n"
+        f"Tomorrow is {_date_label(tomorrow)} ({tomorrow.strftime('%Y-%m-%d')}) "
+        "in this timezone.\n"
+        "Use this for any 'today', 'tomorrow', 'tonight', 'this week', or other "
+        "relative-date reasoning. Do not ask for an exact date just because the "
+        "user used a relative date.\n\n"
+    )
+    return {
+        "role": "user",
+        "content": (
+            "[Context — current date/time, refreshed each turn; not part of "
+            "your instructions]\n" + prompt
+        ),
+    }
+
+
 def current_datetime_context_message(now_utc: Optional[datetime] = None) -> Dict[str, str]:
     """Build the current-date/time context as a standalone chat message.
 

tests/test_scheduler_prompt_cache_time.py

@@ -0,0 +1,136 @@
+"""Regression tests for #4850 — scheduled-task system prompt must not embed
+a minute-level timestamp that busts the Anthropic prompt cache.
+
+Three focused tests:
+1. End-to-end: system prompt is clean; message ordering is [system, datetime
+   user-context, task user-prompt] through the real _run_agent_loop.
+2. Fallback: same ordering when the agent loop raises and task_llm_call_async
+   is used directly.
+3. Helper: current_datetime_context_message_for_tz() renders the correct local
+   time for an explicit IANA timezone, and falls back to UTC for None or invalid.
+"""
+
+from __future__ import annotations
+
+from datetime import datetime, timezone
+from types import SimpleNamespace
+
+
+def _make_task(prompt="run the digest"):
+    return SimpleNamespace(
+        crew_member_id=None, endpoint_url="http://ep/v1", model="m",
+        session_id="s", owner="admin", prompt=prompt,
+        name="job", max_steps=5, character_id=None,
+    )
+
+
+def _patch_scheduler_deps(monkeypatch):
+    monkeypatch.setattr(
+        "src.settings.get_setting",
+        lambda key, default=None: [] if key == "disabled_tools" else default,
+    )
+    monkeypatch.setattr("src.tool_index.get_tool_index", lambda: None)
+
+
+# ---------------------------------------------------------------------------
+# Test 1 — end-to-end: system is clean; agent-loop message ordering is correct
+# ---------------------------------------------------------------------------
+
+async def test_scheduler_agent_loop_path(monkeypatch):
+    """Drive _execute_llm_task end-to-end (real _run_agent_loop, stubbed
+    stream_agent_loop).  Asserts:
+      - system message contains no 'Current time:' prefix
+      - messages[1] is a user-role date/time context block
+      - messages[2] is the task prompt
+    """
+    _patch_scheduler_deps(monkeypatch)
+
+    captured = {}
+
+    async def _stub_stream(**kwargs):
+        captured["messages"] = list(kwargs.get("messages", []))
+        return
+        yield  # async generator
+
+    monkeypatch.setattr("src.agent_loop.stream_agent_loop", _stub_stream)
+    monkeypatch.setattr("src.task_endpoint.resolve_task_candidates", lambda **kw: [])
+
+    from src.task_scheduler import TaskScheduler
+    await TaskScheduler(session_manager=None)._execute_llm_task(_make_task(), db=None)
+
+    msgs = captured.get("messages", [])
+    assert len(msgs) == 3, f"expected 3 messages, got {len(msgs)}"
+    assert msgs[0]["role"] == "system"
+    assert "Current time:" not in msgs[0]["content"]
+    assert msgs[1]["role"] == "user"
+    assert "## Current date and time" in msgs[1]["content"]
+    assert msgs[2]["role"] == "user"
+    assert msgs[2]["content"] == "run the digest"
+
+
+# ---------------------------------------------------------------------------
+# Test 2 — fallback path receives the same datetime context
+# ---------------------------------------------------------------------------
+
+async def test_scheduler_fallback_path(monkeypatch):
+    """When _run_agent_loop raises, task_llm_call_async must receive
+    [system, datetime user-context, task user-prompt] — the same ordering."""
+    _patch_scheduler_deps(monkeypatch)
+
+    captured = {}
+
+    async def _fail(*args, **kwargs):
+        raise RuntimeError("simulated failure")
+
+    async def _capture_call(messages, **kw):
+        captured["messages"] = list(messages)
+        return "fallback"
+
+    import src.task_endpoint as _te
+    monkeypatch.setattr(_te, "task_llm_call_async", _capture_call)
+
+    from src.task_scheduler import TaskScheduler
+    sched = TaskScheduler(session_manager=None)
+    sched._run_agent_loop = _fail
+    await sched._execute_llm_task(_make_task(prompt="send the digest"), db=None)
+
+    msgs = captured.get("messages", [])
+    assert len(msgs) == 3, f"expected 3 messages, got {len(msgs)}"
+    assert msgs[0]["role"] == "system"
+    assert "Current time:" not in msgs[0]["content"]
+    assert msgs[1]["role"] == "user"
+    assert "## Current date and time" in msgs[1]["content"]
+    assert msgs[2]["role"] == "user"
+    assert msgs[2]["content"] == "send the digest"
+
+
+# ---------------------------------------------------------------------------
+# Test 3 — current_datetime_context_message_for_tz() timezone resolution
+# ---------------------------------------------------------------------------
+
+def test_datetime_context_message_for_tz(monkeypatch):
+    """Three cases with a fixed UTC timestamp (2026-06-25 18:00 UTC):
+      - explicit 'America/New_York' → 2:00 PM EDT, UTC-04:00
+      - None                        → UTC fallback: 6:00 PM, UTC+00:00
+      - invalid IANA name           → UTC fallback: same
+    """
+    from src.user_time import current_datetime_context_message_for_tz
+
+    fixed = datetime(2026, 6, 25, 18, 0, tzinfo=timezone.utc)
+
+    # Explicit IANA timezone
+    msg = current_datetime_context_message_for_tz("America/New_York", fixed)
+    assert msg["role"] == "user"
+    assert "America/New_York" in msg["content"]
+    assert "UTC-04:00" in msg["content"]
+    assert "2:00 PM" in msg["content"]
+
+    # None → UTC (preserves old scheduler behaviour for tasks without a crew tz)
+    msg = current_datetime_context_message_for_tz(None, fixed)
+    assert "UTC+00:00" in msg["content"]
+    assert "6:00 PM" in msg["content"]
+
+    # Invalid IANA name → UTC fallback, no exception raised
+    msg = current_datetime_context_message_for_tz("Not/A_Real_Zone", fixed)
+    assert "UTC+00:00" in msg["content"]
+    assert "6:00 PM" in msg["content"]

tests/test_task_shell_tools.py

@@ -111,7 +111,8 @@ async def test_scheduled_task_honors_global_disabled_tools(monkeypatch):
     captured = {}
 
     async def _capture(endpoint_url, model, task, session_id, *,
-                       system_prompt=None, disabled_tools=None, relevant_tools=None):
+                       system_prompt=None, disabled_tools=None, relevant_tools=None,
+                       datetime_context_msg=None):
         captured["disabled_tools"] = disabled_tools
         captured["relevant_tools"] = relevant_tools
         return "done"