Scheduler

The scheduler is the central coordinator. It runs a main loop that checks every subsystem on configurable intervals and dispatches worker threads to process tasks.

Modes

Daemon mode (run_daemon): Long-running process with a WorkerPool. Acquires a file lock on /tmp/istota-scheduler-daemon.lock to prevent duplicate instances. Handles SIGTERM/SIGINT for graceful shutdown. It also starts the persistent AsyncRuntime (see below) explicitly and stops it on shutdown.

Single-pass mode (run_scheduler): Runs all checks once, processes tasks until none remain or a max_tasks limit is hit, then exits. Used for testing and one-off runs. It shares process_one_task (which uses run_coro), so it lazily starts the same persistent runtime, then calls reset_async_runtime() before returning for a clean shutdown.

Persistent asyncio runtime

All Nextcloud Talk I/O runs on one long-lived asyncio loop on a dedicated daemon thread, against one pooled httpx.AsyncClient, instead of a fresh asyncio.run loop + fresh client per call (src/istota/async_runtime.py). This gives TCP/TLS connection reuse to Nextcloud and removes the per-call loop-teardown leak surface.

• AsyncRuntime owns the loop thread; submit(coro, *, timeout=None) bridges sync→async via run_coroutine_threadsafe. stop(timeout=10) cancels in-flight coroutines, runs cleanup hooks (closing the shared client), then stops the loop — cancel-before-aclose so a hook can't close the client out from under a live request.
• run_coro(coro, *, timeout=None) is the workhorse every sync Talk call site uses (run_coro(post_result_to_talk(...)), run_coro(poll_talk_conversations(config)), …). It lazily starts the process-global runtime on first use.
• get_talk_client(config) is a process-global persistent TalkClient singleton; every Talk delivery path pulls from it so they share one connection pool.

Invariant: every TalkClient method invocation must end up on the persistent loop (via run_coro), because the methods issue requests on the loop-bound client. There are no transient TalkClient(config) constructions left in daemon Talk paths. Email delivery stays on asyncio.run (sync SMTP, not httpx).

Main loop

recover_orphaned_tasks_on_startup()  # once, under the flock, before any worker spawns

while not shutdown_requested:
    watchdog.tick()
    pool.dispatch()             # FIRST — spawn workers for users with pending tasks
    check_briefings()           # every briefing_check_interval (60s)
    check_shared_blocks()       # every briefing_check_interval
    check_briefing_triggers()   # every tasks_file_poll_interval (30s; NC-app trigger files)
    check_scheduled_jobs()      # every briefing_check_interval
    _run_sleep_cycles()         # off-thread; polled every briefing_check_interval
    check_travel_timezone()     # off-thread; every TRAVEL_TZ_CHECK_INTERVAL
    poll_emails()               # every email_poll_interval (60s)
    discover_and_organize_shared_files()  # every shared_file_check_interval (120s)
    poll_all_tasks_files()      # every tasks_file_poll_interval (30s)
    run_cleanup_checks()        # every briefing_check_interval
    write_status()              # every 60s
    check_db_health()           # off-thread; every db_health_check_interval (24h)
    _run_db_backup()            # off-thread; every db_backup_interval (24h)
    _maybe_alert_backup_stale() # every tick
    _emit_scheduler_stats()     # every scheduler_stats_interval (60s; 0 disables)
    check_heartbeats()          # every heartbeat_check_interval (60s) — last gated check
    sleep(poll_interval)        # 2s, sub-ticking pool.dispatch() every dispatch_interval

pool.dispatch() runs first, before the clock is even read — pending-task latency is the thing the loop is optimizing for, and every interval check below it is a potential delay.

Talk polling runs in a separate daemon thread, started at scheduler launch.

Off-thread periodic checks

Four checks can outlast a tick — the DB health sweep and the DB backup snapshot both walk every per-user DB (the backup writing to the rclone FUSE mount, where latency is unbounded), the nightly sleep cycles make synchronous per-user LLM calls, and the travel-timezone check opens a per-user location.db and may send a notification. Run inline they blocked pool.dispatch() for their whole duration. _spawn_background_check(name, fn, inflight, *, overlap_expected=False) puts each on a short-lived bgcheck-<name> daemon thread, skipping the tick when the previous run under the same name is still alive, so a wedged sweep cannot stack one thread per tick. overlap_expected=True (sleep cycles, travel timezone) demotes that skip log to DEBUG, because for those two a pass outliving the poll interval is the normal case rather than a symptom. Exceptions are contained and logged. The interval clocks advance at spawn time (fixed cadence; the in-flight guard prevents overlap), while the staleness alert reads the persisted last-run clock, which only advances on a durable OK run.

Because none of the known-long checks runs on the loop thread any more, there are no LoopWatchdog.suspended() call sites left — the stall watchdog covers the whole loop.

Startup orphan recovery

The heartbeat-based stuck-task reclaim infers a dead worker and takes up to worker_stuck_minutes to do it. A scheduler restart is deterministic instead: the daemon holds a singleton flock, so the moment a fresh instance boots, every running / locked row belongs to the dead instance. recover_orphaned_tasks_on_startup runs once under the flock, before any worker spawns, and resolves each orphan in priority order — cancel_requested → cancelled; retries exhausted, too old, or an inline-only source (REPL) → failed; otherwise back to pending with attempt_count bumped and every liveness column cleared. Terminal outcomes emit a terminal event frame so a watching web client gets closure instead of a hung spinner; released orphans emit nothing, since the re-run streams its own task_started. pending_confirmation rows are left alone.

Worker pool

WorkerPool manages concurrent UserWorker threads with three-tier concurrency control:

Instance-level caps: max_foreground_workers (default 5) and max_background_workers (default 3) limit total concurrent workers by queue type. Dispatch is two-phase: foreground first, then background.

Per-user limits: user_max_foreground_workers (default 2) and user_max_background_workers (default 1) cap how many workers a single user can have. Individual users can override these in their config.

Per-channel gate: Before creating a task, the Talk poller checks if an active foreground task exists for the conversation. If so, it queues the message but sends "Still working on a previous request" as an immediate response.

Workers are keyed by (user_id, queue_type, slot). Each UserWorker is a thread that processes tasks serially, exiting after worker_idle_timeout (10s) of no tasks. Thread safety: fresh DB connections per call, new asyncio.run() event loop per worker, threading.Lock on the workers dict.

Task claiming

claim_task() uses atomic UPDATE...RETURNING with stale lock detection:

1. Fail old stale locked tasks (created > max_retry_age, locked > 30 min)
2. Release recent stale locks for retry
3. Fail old stuck running tasks
4. Release recent stuck running tasks for retry
5. Claim next pending: ORDER BY priority DESC, created_at ASC

The claim sets status='locked', locked_at=now, locked_by=worker_id atomically.

Stuck-running detection by worker liveness (ISSUE-112)

Steps 3–5 (and the standalone fail_stuck_locked_running_tasks() maintenance pass) share _STUCK_RUNNING_PREDICATE, which decides "stuck" by worker liveness, not raw runtime. A running task is stuck when its last_heartbeat has been silent longer than worker_stuck_minutes (default 10); when no heartbeat was ever recorded it falls back to started_at older than task_timeout_minutes + grace. The running worker pings last_heartbeat every worker_heartbeat_seconds (default 60) via the _task_heartbeat context manager (db.touch_task_heartbeat), so a slow-but-alive worker — notably the in-process native brain, which has no killable PID — is never reclaimed, while a crashed worker is recovered in minutes. (This is distinct from the health-check heartbeat system in heartbeat.py.)

claim_task and every other Task-returning helper (get_task, get_pending_confirmation*, get_reply_parent_task, get_stale_pending_tasks, get_completed_*_since) route their SELECT/RETURNING through a single _TASK_COLUMNS constant in db.py. Adding a column means editing one place; missing columns now raise IndexError rather than silently returning None (the failure mode that masked the brief 027eb1a regression where task.skill came back unset and module-poller rows fell through to the LLM path with an empty prompt).

Task dispatch

process_one_task decides between three execution paths based on the task's columns:

Task shape	Dispatcher	Notes
task.skill set	_execute_skill_task()	python -m istota.skills.<skill> subprocess. Trusted env via build_skill_env(list(skill_index), skill_index, ctx) over the full index, so co-declared vars (e.g. NC_URL declared on both files and nextcloud) reach the subprocess. No proxy split.
task.command set	_execute_command_task()	Shell command. Admin-gated (non-admin tasks refused at runtime + dropped at sync time). Same trusted env resolver as skill-tasks. JSON {"status":"error","error":"…"} envelopes on stdout are detected and surfaced as failures even when returncode is 0.
neither	LLM path via the brain	Default; runs through execute_task

Auto-seeded _module.feeds.run_scheduled / _module.money.run_scheduled rows dispatch as skill-tasks. _purge_obsolete_skill_jobs removes rows whose skill name is no longer in the index.

Task processing

process_one_task() handles the full lifecycle:

1. Claim a task (with optional user_id filter)
2. Update status to running
3. Get user resources, send Talk acknowledgment, download attachments
4. Call execute_task() -> (success, result, actions_taken, execution_trace)
5. On success:
   • Check for malformed output (leaked tool-call XML) -> reclassify as failure
   • Check for confirmation request (regex pattern)
   • Update to completed
   • Index conversation for memory search
   • Deliver results
6. On failure:
   • Check cancellation
   • Retry with exponential backoff if attempts remain (1, 4, 16 min delays)
   • Mark permanently failed after max attempts

Retry logic

Failed tasks retry with exponential backoff: 1 min, 4 min, 16 min (up to max_attempts, default 3). Transient API errors (5xx, 429) get 3 fast retries with 5s delay before counting against task attempts.

Shutdown collateral is not a failure. Under systemd's default KillMode=control-group, a systemctl restart (the auto-update cron issues one per commit) SIGTERMs the whole cgroup, killing an in-flight task's model subprocess while the daemon shuts down gracefully — so the surviving worker recorded the corpse as a failure, permanently so on a final attempt. When _shutdown_requested is set and the failure text carries a signal-termination marker, the task goes back to pending via db.release_task_for_restart with no attempt charged and no backoff, and its deferred-op files are purged; fail_ancient_pending_tasks remains the bound. The client sees a "Scheduler restarting…" progress notice rather than a terminal frame. The unit template also sets KillMode=mixed, which converts the same event into the startup orphan-recovery path — this branch is the belt-and-braces half, and the one that ships via auto-update since unit files need an Ansible run.

A related invariant: worker_pid is cleared on every transition out of running (completed, failed, cancelled, pending-retry, restart-release, orphan recovery). It used to survive a failed attempt, so a retry row could carry a dead attempt's PID — and both cancel paths (!stop, the web cancel endpoint) signal whatever the row holds, which would eventually land on an unrelated recycled PID.

Task event streaming

One persistent, typed event stream per task feeds every output surface. process_one_task builds an EventWriter (events.py) per brain-path task and subscribes the in-process consumers (TalkEventSubscriber, LogChannelSubscriber, PushNotificationSubscriber) before passing the writer to execute_task(event_writer=…). The executor adapts the brain's StreamEvent stream into TaskEvents, persisted to the task_events table (WAL, shared scheduler ⇄ web). When the task reaches a non-retry terminal state the scheduler emits the terminal event (confirmation / result / cancelled / error + done) and calls writer.finish().

Retry continuity: on a retry-eligible failure the event log is kept (not wiped). The retry branch emits a progress_text "⏳ Attempt failed — retrying in N min…" notice, and the next attempt's EventWriter resumes seq from db.get_max_task_event_seq so it stays monotonic across attempts and a watching web client survives the retry instead of hanging on "Working…". The SSE / snapshot endpoints synthesize a terminal frame from the task row (web_app._synthetic_terminal_events) for any terminal-without-done gap (e.g. a crash that skipped finish()).

Config under [scheduler]: progress_show_tool_use, progress_show_text, event_log_enabled, stream_text_gate_chars, push_notification_threshold_seconds, push_notification_sources.

Delivery routing

Where a task's result goes is resolved by transport.routing.resolve_delivery_plan(config, task, registry), which turns a task into an ordered, deduplicated, channel-resolved list of destinations. Precedence: explicit output_target > reply-to-origin (interactive source types) > source-type default > drop. process_one_task builds the plan once and fans out to every push destination; stream destinations (REPL, web) contribute no push work — the task_events log is the delivery. Separately, a per-user purpose-keyed routing table (UserConfig.routing, purposes reply/alert/log/briefing/notification) routes notifications via notifications.send_notification(..., purpose=…). See the Transport abstraction and .claude/rules/transport.md.

Deferred DB operations

With the bubblewrap sandbox, no database is reachable inside the subprocess at all — the framework DB directory and the per-user module-DB root are covered by empty, read-only tmpfs masks applied as the last mount operations. Claude and skill CLIs write JSON files to a writable temp dir instead. The scheduler processes these after successful completion. The handlers and the file envelope helper live in scheduler_deferred.py (extracted from scheduler.py for size and testability; scheduler.py keeps a re-export shim so from istota.scheduler import _process_deferred_* still works).

File	Handler	Purpose
task_{id}_subtasks.json	_process_deferred_subtasks	Subtask creation (admin-only, depth- and rate-capped)
task_{id}_tracked_transactions.json	_process_deferred_tracking	Transaction dedup tracking
task_{id}_sent_emails.json	_process_deferred_sent_emails	Outbound email tracking for emissary thread matching
task_{id}_kv_ops.json	_process_deferred_kv_ops	KV store set/delete operations
task_{id}_kg_ops.json	_process_deferred_kg_ops	Knowledge-graph fact add/invalidate/delete (per-op commit)
task_{id}_user_alerts.json	_process_deferred_user_alerts	Alerts/notifications for the user's alerts channel
task_{id}_health_ops.json	_process_deferred_health_ops	Health module inserts/updates replayed against the per-user health.db
task_{id}_email_output.json	_deliver_deferred_email_output	Structured email reply (preferred over the legacy stdout-JSON parser)
task_{id}_garmin_import.json	_process_deferred_garmin_import	Garmin Connect sync requested from inside the sandbox

_load_deferred_json(user_temp_dir, task_id, suffix, expected_type=...) is the shared envelope helper: builds the path, exists-checks, parses JSON, validates the top-level shape (list or dict), and warns + unlinks on a malformed file. Each handler then runs its own business logic and unlinks at the call site so per-handler invariants (admin gate, depth gate, KG per-op commit) read cleanly.

_purge_deferred_files_for_retry clears the slate when a task is re-claimed after a crash, a restart, or a retry, so a non-idempotent op like a KG invalidate isn't replayed twice across attempts. _warn_unconsumed_deferred_files scans the user temp dir after the drain phase and logs WARN for files missing the task_ prefix or carrying an unknown suffix; the misnamed file is left on disk for inspection.

One suffix is recognized but never purged: task_{id}_health_op_failures.json, in _KNOWN_ARTIFACT_SUFFIXES. It is written by a handler rather than read by one — the rows a health op lost mid-batch, preserved so an operator can recover them after the task settles.

Identity fields (user_id, conversation_token) always come from the task, not from the JSON, to prevent spoofing.

Cleanup

Runs every briefing_check_interval:

• Cancel stale confirmations after 120 min, notify user
• Recover stuck locked/running tasks (mirrors the claim_task recovery, for rows no claim happens to touch)
• Log warnings for tasks pending longer than 30 min
• Auto-fail tasks pending longer than stale_pending_fail_hours (2)
• Delete completed tasks older than task_retention_days (7)
• Prune processed_emails rows older than processed_email_retention_days (90, floored at email_retention_days + 1), excluding rows the stored transcript still references
• Prune the message_deletions ledger (fixed 30 days) — it exists only to tell a reconnecting client what vanished while it was away
• Delete processed emails from IMAP older than email_retention_days (7), via one server-side BEFORE search
• Trim the Talk message cache
• Delete old temp files
• Delete location pings older than location_ping_retention_days (365) from each per-user location.db
• Reconcile visits from pings (batch cleanup of visit state-machine drift)
• Delete old Claude session logs

Poller intervals

Poller	Default interval	Config key
Task queue	2s	poll_interval
Pending-task dispatch sub-tick	0.5s	dispatch_interval
Talk conversations	10s	talk_poll_interval
Email (IMAP)	60s	email_poll_interval
Briefings/jobs/sleep/cleanup	60s	briefing_check_interval
TASKS.md files	30s	tasks_file_poll_interval
Shared files	120s	shared_file_check_interval
Heartbeats	60s	heartbeat_check_interval
SQLite health (quick_check + self-heal REINDEX)	86400s (24h)	db_health_check_interval

dispatch_interval decouples cold pending-task pickup latency from the interval-gated checks: the main loop runs pool.dispatch() on this sub-tick cadence without re-running the per-subsystem checks (0 or ≥ poll_interval = legacy one-dispatch-per-tick). cron_max_staleness_minutes (default 60) is the insertion-time staleness gate for check_scheduled_jobs / check_briefings — after a long outage it skips the catch-up insert and resumes from the next future fire, suppressing thundering-herd catch-up.