Temporal Reasoning & Scheduling
Synchronous agents are simpler to implement. They process a request, execute steps as fast as they can, and return a result. Every action happens now, so their relationship with time consists entirely of tool-call latency.
Real-world tasks are soaked in time. "Remind me to follow up on Friday." "Run this report every Monday at 8 AM." "Start the data migration after the backup completes, but only during the maintenance window." "This task has three phases — phase one must finish by end of day, phase two depends on an external review that takes 24–48 hours, and phase three must complete before the quarterly deadline." Temporal reasoning expands an agent beyond a single synchronous execution by adding deadlines, causal ordering, and future scheduling.
This article covers how to give agents a relationship with time: representing temporal concepts in tool schemas, planning under deadlines, managing dependencies between time-bound tasks, and scheduling deferred execution.
Why Time Is Hard for Language Models #
Language models derive their sense of time entirely from the context window: timestamps in messages, dates in documents, and explicit instructions about the current time. Applications must supply and refresh that temporal context.
This creates several failure modes:
Stale context. If the system prompt says "Today is July 15" and the conversation spans three days, the model still thinks it is July 15. Refreshing the context advances the model's current time.
Duration blindness. Models are poor at estimating how long things take. "This API call usually returns in 200ms" and "this human review takes 2 business days" are qualitatively different, so the prompt must encode their expected durations explicitly.
Calendar reasoning failures. "Schedule a meeting for the third business day after next Tuesday" requires calendar arithmetic that models get wrong surprisingly often — especially across month boundaries, holidays, and time zones.
Temporal ordering confusion. In a long conversation with many events, models lose track of what happened before what. "The deploy happened before the rollback" versus "the rollback happened before the deploy" can get confused when both events appear in the context.
What the agent sees:
System: "Current time: 2026-07-31T10:00:00Z"
User: "Schedule a follow-up 3 business days from now"
What the agent needs to compute:
July 31 (Thu) → Aug 1 (Fri) → Aug 4 (Mon) → Aug 5 (Tue)
Answer: August 5, 2026
What the agent often gets wrong:
"3 days from July 31 is August 3" (ignores weekends)
"3 business days is August 1" (counts today)
"August 5 is a Wednesday" (wrong day of week)
The practical fix is to give the model tools that handle calendar math correctly.
Time-Aware Tools #
The first step is providing tools that query time and perform the arithmetic for the agent.
import json
from datetime import datetime, timedelta
from zoneinfo import ZoneInfo
def parse_datetime(value: str, timezone: str = "UTC") -> datetime:
"""Parse ISO 8601 and normalize it to the requested timezone."""
tz = ZoneInfo(timezone)
parsed = datetime.fromisoformat(value)
if parsed.tzinfo is None:
return parsed.replace(tzinfo=tz)
return parsed.astimezone(tz)
def get_current_time(timezone: str = "UTC") -> str:
"""Return the current date and time in the specified timezone."""
tz = ZoneInfo(timezone)
now = datetime.now(tz)
return now.isoformat()
def add_business_days(start_date: str, days: int, timezone: str = "UTC") -> str:
"""
Add N business days to a date, skipping weekends.
Returns the resulting date as ISO 8601.
"""
if days < 0:
raise ValueError("days must be non-negative")
current = parse_datetime(start_date, timezone)
added = 0
while added < days:
current += timedelta(days=1)
if current.weekday() < 5: # Monday=0, Friday=4
added += 1
return current.date().isoformat()
def time_until(target: str, timezone: str = "UTC") -> str:
"""
Calculate the duration between now and a target datetime.
Returns a human-readable duration string.
"""
tz = ZoneInfo(timezone)
now = datetime.now(tz)
target_dt = parse_datetime(target, timezone)
delta = target_dt - now
total_seconds = int(abs(delta.total_seconds()))
days, remainder = divmod(total_seconds, 86400)
hours, remainder = divmod(remainder, 3600)
minutes = remainder // 60
direction = "ago" if delta.total_seconds() < 0 else "remaining"
return f"{days} days, {hours} hours, {minutes} minutes {direction}"
def is_within_window(
check_time: str,
window_start: str,
window_end: str,
) -> str:
"""Check if a datetime falls within a time window."""
check = parse_datetime(check_time)
start = parse_datetime(window_start)
end = parse_datetime(window_end)
if end < start:
raise ValueError("window_end must be at or after window_start")
within = start <= check <= end
return json.dumps({"within_window": within})
With these tools, the agent delegates "3 business days from July 31" to add_business_days("2026-07-31", 3) and gets back "2026-08-05". The calendar math is correct by construction.
Injecting Time into Context #
The system prompt should always include the current time, and it should be refreshed on every turn of a multi-turn conversation:
from datetime import datetime
from zoneinfo import ZoneInfo
def build_time_context(timezone: str = "UTC") -> str:
tz = ZoneInfo(timezone)
now = datetime.now(tz)
return (
f"Current time: {now.isoformat()}\n"
f"Day of week: {now.strftime('%A')}\n"
f"Timezone: {timezone}"
)
def build_system_prompt(base_prompt: str, timezone: str = "UTC") -> str:
time_context = build_time_context(timezone)
return f"{time_context}\n\n{base_prompt}"
Refreshing the time context on each turn is important for long-running agents that span hours or days. A durable agent that resumes from a checkpoint must replace the checkpoint timestamp with the current time.
Deadline-Aware Planning #
When a task has a deadline, the agent's planning must account for time constraints. This means estimating durations for sub-tasks and checking whether the plan fits within the available time before starting execution.
from dataclasses import dataclass
@dataclass
class TimeBoundTask:
task_id: str
description: str
estimated_duration_minutes: float
depends_on: list[str]
deadline: str | None = None # ISO 8601 datetime
started_at: str | None = None
completed_at: str | None = None
@dataclass
class ScheduleValidation:
feasible: bool
total_estimated_minutes: float
critical_path_minutes: float
slack_minutes: float
warnings: list[str]
def validate_schedule(
tasks: list[TimeBoundTask],
deadline: str,
start_time: str | None = None,
) -> ScheduleValidation:
"""
Check whether a set of tasks with dependencies
can complete before a deadline.
"""
end = parse_datetime(deadline)
start = (
parse_datetime(start_time)
if start_time
else datetime.now(ZoneInfo("UTC"))
)
available_minutes = (end - start).total_seconds() / 60
task_map = {t.task_id: t for t in tasks}
if len(task_map) != len(tasks):
raise ValueError("task_id values must be unique")
if any(t.estimated_duration_minutes < 0 for t in tasks):
raise ValueError("estimated durations must be non-negative")
warnings = []
# Compute critical path with memoized depth-first traversal.
finish_times: dict[str, float] = {}
visiting: set[str] = set()
def get_earliest_finish(task_id: str) -> float:
if task_id in finish_times:
return finish_times[task_id]
if task_id in visiting:
raise ValueError(f"Dependency cycle detected at task '{task_id}'")
if task_id not in task_map:
raise ValueError(f"Unknown task dependency: '{task_id}'")
visiting.add(task_id)
task = task_map[task_id]
earliest_start = max(
(get_earliest_finish(dep) for dep in task.depends_on),
default=0.0,
)
finish = earliest_start + task.estimated_duration_minutes
visiting.remove(task_id)
finish_times[task_id] = finish
return finish
for t in tasks:
get_earliest_finish(t.task_id)
critical_path = max(finish_times.values()) if finish_times else 0.0
total = sum(t.estimated_duration_minutes for t in tasks)
slack = available_minutes - critical_path
if slack < 0:
warnings.append(
f"Critical path ({critical_path:.0f} min) exceeds "
f"available time ({available_minutes:.0f} min) by {-slack:.0f} min"
)
if 0 <= slack < critical_path * 0.1:
warnings.append("Less than 10% schedule slack — high risk of missing deadline")
feasible = slack >= 0
for task in tasks:
if task.deadline:
task_end = parse_datetime(task.deadline)
task_available = (task_end - start).total_seconds() / 60
if finish_times[task.task_id] > task_available:
feasible = False
warnings.append(
f"Task '{task.task_id}' exceeds its own deadline"
)
return ScheduleValidation(
feasible=feasible,
total_estimated_minutes=total,
critical_path_minutes=critical_path,
slack_minutes=slack,
warnings=warnings,
)
The agent's planning step produces a list of TimeBoundTask objects with duration estimates and dependencies. The validate_schedule function checks feasibility before execution starts. If the plan is infeasible — the critical path exceeds the deadline — the agent can simplify the plan, parallelize independent tasks, or negotiate a deadline extension with the user.
Adaptive Replanning Under Time Pressure #
Plans that were feasible at the start can become infeasible during execution. A tool call that was estimated at 5 minutes takes 20. An external dependency that was expected to resolve in an hour is still pending. The agent needs to detect when it is falling behind and adjust.
def check_schedule_health(
tasks: list[TimeBoundTask],
deadline: str,
) -> dict:
"""
Check whether in-progress execution is still on track.
Call this after each task completes.
"""
end = parse_datetime(deadline)
now = datetime.now(ZoneInfo("UTC"))
remaining_minutes = (end - now).total_seconds() / 60
completed = [t for t in tasks if t.completed_at]
pending = [t for t in tasks if not t.completed_at]
pending_ids = {t.task_id for t in pending}
remaining_tasks = [
TimeBoundTask(
task_id=t.task_id,
description=t.description,
estimated_duration_minutes=t.estimated_duration_minutes,
depends_on=[dep for dep in t.depends_on if dep in pending_ids],
deadline=t.deadline,
)
for t in pending
]
validation = validate_schedule(
remaining_tasks,
deadline=deadline,
start_time=now.isoformat(),
)
remaining_work = sum(t.estimated_duration_minutes for t in pending)
status = "on_track"
if not validation.feasible:
status = "behind_schedule"
elif validation.slack_minutes < validation.critical_path_minutes * 0.2:
status = "at_risk"
return {
"status": status,
"remaining_minutes": remaining_minutes,
"remaining_work_estimate": remaining_work,
"remaining_critical_path": validation.critical_path_minutes,
"completed_tasks": len(completed),
"pending_tasks": len(pending),
}
The agent calls check_schedule_health after completing each sub-task. When the status shifts to at_risk or behind_schedule, the model can reason about trade-offs: drop optional sub-tasks, reduce depth on remaining research, switch to a faster (cheaper) model for remaining steps, or alert the user that the deadline is at risk.
Temporal Dependencies #
Some tasks extend data dependencies such as "task B needs task A's output" with wall-clock requirements such as "task B can start at 9 AM" or "task B must wait at least 2 hours after task A." These are temporal constraints.
from dataclasses import dataclass, field
from enum import Enum
class ConstraintType(Enum):
AFTER_TASK = "after_task" # Start after another task finishes
NOT_BEFORE = "not_before" # Earliest allowed start time
NOT_AFTER = "not_after" # Latest allowed start time
DELAY_AFTER = "delay_after" # Wait N minutes after a task
WITHIN_WINDOW = "within_window" # Must execute during a time window
@dataclass
class TemporalConstraint:
constraint_type: ConstraintType
reference_task_id: str | None = None
reference_time: str | None = None # ISO 8601
delay_minutes: float | None = None
window_start: str | None = None # ISO 8601
window_end: str | None = None # ISO 8601
@dataclass
class ScheduledTask:
task_id: str
description: str
estimated_duration_minutes: float
constraints: list[TemporalConstraint] = field(default_factory=list)
scheduled_start: str | None = None
scheduled_end: str | None = None
scheduling_error: str | None = None
def resolve_schedule(
tasks: list[ScheduledTask],
completed: dict[str, str], # task_id -> completion ISO time
) -> list[ScheduledTask]:
"""
Assign scheduled_start times to tasks based on
their temporal constraints and current completion state.
"""
now = datetime.now(ZoneInfo("UTC"))
for task in tasks:
if task.task_id in completed:
continue
earliest = now
latest_start = None
latest_end = None
for c in task.constraints:
if c.constraint_type == ConstraintType.AFTER_TASK:
if c.reference_task_id not in completed:
task.scheduling_error = "Waiting for a dependency"
earliest = None
break
dep_done = parse_datetime(completed[c.reference_task_id])
earliest = max(earliest, dep_done)
elif c.constraint_type == ConstraintType.DELAY_AFTER:
if c.reference_task_id not in completed:
task.scheduling_error = "Waiting for a dependency"
earliest = None
break
if c.delay_minutes is None:
raise ValueError("DELAY_AFTER requires delay_minutes")
dep_done = parse_datetime(completed[c.reference_task_id])
delayed = dep_done + timedelta(minutes=c.delay_minutes)
earliest = max(earliest, delayed)
elif c.constraint_type == ConstraintType.NOT_BEFORE:
if c.reference_time is None:
raise ValueError("NOT_BEFORE requires reference_time")
not_before = parse_datetime(c.reference_time)
earliest = max(earliest, not_before)
elif c.constraint_type == ConstraintType.NOT_AFTER:
if c.reference_time is None:
raise ValueError("NOT_AFTER requires reference_time")
latest_start = parse_datetime(c.reference_time)
elif c.constraint_type == ConstraintType.WITHIN_WINDOW:
if c.window_start is None or c.window_end is None:
raise ValueError("WITHIN_WINDOW requires window_start and window_end")
w_start = parse_datetime(c.window_start)
w_end = parse_datetime(c.window_end)
earliest = max(earliest, w_start)
latest_end = w_end
if earliest is not None:
if latest_start and earliest > latest_start:
task.scheduling_error = "Latest start time has passed"
continue
task.scheduled_start = earliest.isoformat()
end = earliest + timedelta(minutes=task.estimated_duration_minutes)
if latest_end and end > latest_end:
task.scheduling_error = "Task exceeds its execution window"
task.scheduled_start = None
continue
task.scheduled_end = end.isoformat()
return tasks
The scheduler resolves constraints incrementally as tasks complete. When the agent finishes a task and records its completion time, the scheduler re-evaluates which pending tasks are now eligible to start. Tasks with unmet dependencies remain unscheduled. Tasks with delay constraints are scheduled for the future.
Deferred and Scheduled Execution #
Some tasks wait for a specific future trigger. "Send this reminder at 9 AM tomorrow." "Run the compliance check every Friday." "Start the migration during the 2 AM maintenance window."
This requires an execution layer that accepts scheduled work and triggers it at the right time. The agent produces a schedule entry; the runtime executes it later.
from dataclasses import dataclass, field
from typing import Any
@dataclass
class ScheduledJob:
job_id: str
agent_id: str
trigger_time: str # ISO 8601
task: str
context: dict[str, Any]
recurrence: str | None = None # "daily" or "weekly"
max_retries: int = 3
created_at: str = field(
default_factory=lambda: datetime.now(ZoneInfo("UTC")).isoformat()
)
class JobScheduler:
"""
Persistent job scheduler that the agent can write to.
A separate worker process polls and executes due jobs.
"""
def __init__(self, store: "JobStore"):
self.store = store
async def schedule(self, job: ScheduledJob) -> str:
await self.store.save(job)
return f"Scheduled job '{job.job_id}' for {job.trigger_time}"
async def get_due_jobs(self) -> list[ScheduledJob]:
now = datetime.now(ZoneInfo("UTC")).isoformat()
return await self.store.query_due(before=now)
async def mark_complete(self, job_id: str) -> None:
job = await self.store.get(job_id)
if job is None:
return
if job and job.recurrence:
next_time = self._compute_next(job)
job.trigger_time = next_time
await self.store.save(job)
else:
await self.store.delete(job_id)
def _compute_next(self, job: ScheduledJob) -> str:
current = datetime.fromisoformat(job.trigger_time)
if job.recurrence == "daily":
return (current + timedelta(days=1)).isoformat()
if job.recurrence == "weekly":
return (current + timedelta(weeks=1)).isoformat()
raise ValueError(f"Unsupported recurrence: {job.recurrence!r}")
The agent exposes schedule creation as a tool:
async def schedule_task(
description: str,
trigger_time: str,
recurrence: str | None = None,
) -> str:
"""
Schedule a task for future execution.
trigger_time: ISO 8601 datetime.
recurrence: 'daily', 'weekly', or None for one-time.
"""
job = ScheduledJob(
job_id=generate_id(),
agent_id=current_agent_id,
trigger_time=trigger_time,
task=description,
context=get_current_context(),
recurrence=recurrence,
)
return await scheduler.schedule(job)
When the model decides "this task should run at 9 AM tomorrow," it calls the scheduling tool with the computed trigger time. The scheduler persists the job, and a background worker picks it up when the time arrives. The worker rehydrates the agent with the saved context and executes the task — the same pattern as resuming a durable agent from a checkpoint.
Time-Bounded Execution #
Some tasks must finish within a time limit. A customer-facing agent that takes 60 seconds to respond has failed even if the answer is correct. A data pipeline step that runs past its maintenance window must stop, regardless of progress.
import asyncio
class TimeBoundedExecution:
"""
Run an agent loop with a hard time limit.
On timeout, return partial results rather than nothing.
"""
def __init__(self, timeout_seconds: float):
self.timeout_seconds = timeout_seconds
async def run(
self,
agent_fn,
task: str,
on_timeout: str = "partial",
) -> dict:
best_result = {"result": None, "partial": True, "steps": 0}
async def bounded_agent():
nonlocal best_result
async for step_result in agent_fn(task):
best_result = {
"result": step_result["output"],
"partial": not step_result.get("complete", False),
"steps": step_result["step"],
}
if step_result.get("complete"):
return best_result
return best_result
try:
result = await asyncio.wait_for(
bounded_agent(),
timeout=self.timeout_seconds,
)
return result
except asyncio.TimeoutError:
if on_timeout == "partial":
return {
**best_result,
"timed_out": True,
"timeout_seconds": self.timeout_seconds,
}
return {
"result": None,
"timed_out": True,
"error": f"Exceeded {self.timeout_seconds}s time limit",
}
The key design choice: preserve partial results on timeout. An agent that spent 55 out of 60 seconds researching and found four of five answers should return those four. The caller — whether a user, a pipeline, or an outer agent — can decide whether the partial result is good enough or needs a follow-up.
Budget-Aware Step Selection #
When an agent knows its time budget, it can make smarter decisions about what to do with the remaining time.
def build_time_budget_context(
deadline: str,
steps_completed: int,
avg_step_seconds: float,
) -> str:
"""
Generate a context block that tells the model
how much time it has left and how to spend it.
"""
end = datetime.fromisoformat(deadline)
now = datetime.now(end.tzinfo)
remaining = max(0.0, (end - now).total_seconds())
estimated_steps_remaining = int(remaining / avg_step_seconds) if avg_step_seconds > 0 else 0
if remaining < 30:
urgency = "CRITICAL: Less than 30 seconds remaining. Produce final answer now."
elif remaining < 120:
urgency = "LOW TIME: ~2 minutes remaining. Wrap up current line of investigation."
else:
urgency = f"On track: ~{remaining / 60:.0f} minutes remaining."
return (
f"Time budget: {urgency}\n"
f"Steps completed: {steps_completed}\n"
f"Estimated steps remaining: {estimated_steps_remaining}\n"
f"Average step duration: {avg_step_seconds:.1f}s"
)
Inject this context into the system prompt on every turn. When the agent sees "CRITICAL: Less than 30 seconds remaining. Produce final answer now," it stops exploring and synthesizes what it has. This graceful deadline produces a coherent partial result before the hard timeout.
Recurring Tasks and Cron Agents #
Schedules trigger some agent work directly: daily summaries, weekly reports, periodic health checks, and automated monitoring. These are cron agents: agents that activate on a time trigger, perform their task, and go dormant until the next trigger.
┌──────────────────────────────────────────────────────┐
│ Cron Agent Lifecycle │
│ │
│ ┌────────┐ trigger ┌──────────┐ complete │
│ │Dormant │────────────▶│ Running │──────────┐ │
│ │ │◀────────────│ │ │ │
│ └────────┘ schedule └──────────┘ │ │
│ ▲ next run │ │ │
│ │ │ on error │ │
│ │ ┌────▼─────┐ │ │
│ └──────────────────│ Retry/ │──────────┘ │
│ after backoff │ Backoff │ │
│ └──────────┘ │
└──────────────────────────────────────────────────────┘
@dataclass
class CronAgent:
agent_id: str
schedule: str # cron expression or "daily", "weekly"
task_template: str
tools: list
system_prompt: str
last_run: str | None = None
last_result: str | None = None
async def run_cron_cycle(
agent_config: CronAgent,
model: "ModelClient",
) -> dict:
"""
Execute one scheduler-triggered cycle and retain temporal memory.
"""
# Build context with time since last run
context_parts = [f"Current time: {datetime.now().isoformat()}"]
if agent_config.last_run:
context_parts.append(f"Last run: {agent_config.last_run}")
if agent_config.last_result:
context_parts.append(f"Previous result summary: {agent_config.last_result[:500]}")
messages = [{"role": "user", "content": (
f"{chr(10).join(context_parts)}\n\nTask: {agent_config.task_template}"
)}]
result = await run_agent_loop(
model=model,
system_prompt=agent_config.system_prompt,
tools=agent_config.tools,
messages=messages,
)
agent_config.last_run = datetime.now().isoformat()
agent_config.last_result = result.result
return {"result": result.result, "schedule": agent_config.schedule}
Cron agents benefit from temporal memory — each run knows when it last ran and what it found. A weekly security scan that remembers last week's findings can highlight changes such as "3 new vulnerabilities since last scan." This is where temporal reasoning and memory intersect: the agent's memory is time-indexed, and the current run's value comes partly from its awareness of what happened in previous runs.
Conclusion #
Temporal reasoning spans a set of design decisions at every layer of the agent stack:
- Give agents time tools. Models are unreliable at calendar arithmetic. Tools that compute business days, time differences, and window checks are correct by construction.
- Inject fresh time context on every turn. Stale timestamps in the system prompt cause cascading errors in long-running or resumed agents.
- Validate schedules before execution. Compute the critical path, check it against the deadline, and flag plans that exceed the available time before execution begins.
- Monitor schedule health during execution. Replanning when falling behind is cheaper than discovering the deadline was missed after the fact.
- Return partial results on timeout. An agent approaching a time limit should hand back its accumulated work. Let the caller decide whether the partial result is sufficient.
- Budget-aware prompting changes agent behavior. Telling the model how much time remains lets it make rational trade-offs between depth and completion.
- Deferred execution needs a persistent scheduler. Agents that schedule future work need a durable job store and a worker that triggers execution at the right time — the same checkpoint-and-resume pattern used by durable agents.
- Cron agents are time-triggered loops. Recurring tasks benefit from temporal memory that tracks both changes between runs and the current state.