fix(orchestrator): record the terminal states that were never written

Two defects with one shape: an execution reaches a terminal condition and the
orchestrator fails to write it down, so the system's own record disagrees with
what happened. Neither produced an error. Both produced silence.

T-2 -- a restart mid-task unscheduled that task forever.

get_tasks_for_minute excludes any task holding a task_executions row with
status='running'. The row is written before the executor runs and updated
after, so a process dying in between left it 'running' permanently, and the
task was then excluded from every future minute with no error, no alarm and no
log line. It did not fail; it went quiet.

test_example_task had held such a row since 2025-12-07 -- 5916 hours. It is
disabled, so nothing was broken by that instance; the mechanism is the point,
and the exposure is daily, because Watchtower restarts this container at 4 AM
while the config backup starts at 03:05 and runs ~21 minutes.

Startup now reconciles them, where the reasoning is sound by construction: this
process has just begun, so nothing it can see is genuinely running.

Marked 'orphaned', not 'failed'. When the process dies mid-task the work may
well have completed -- a backup that finished and never got to update its row
is indistinguishable from one that died halfway -- and 'failed' would assert an
outcome nobody observed. Same error as the health-report diagnostic fixed in
18be804: naming a cause you did not witness.

Not extended to a duration-based sweep. While this process lives, execute_task's
finally clause always closes the row, so a stale row implies a dead owner. A
time-based rule would have to tell a slow task from a dead one, and getting that
wrong closes the record of a task still working.

T-3 -- the 'timeout' status was unreachable.

execute_task has an `except asyncio.TimeoutError` branch that records
status='timeout'. It could never run: _run_executor wrapped the awaited call in
`except Exception`, and since 3.11 asyncio.TimeoutError IS the builtin
TimeoutError (OSError -> Exception), so the broad handler caught it first and
converted it to an ordinary error tuple. Confirmed in the deployed runtime and
against the history -- 18,785 executions since 2025-12-07, of which 'timeout'
rows: zero. Every timeout in eight months was filed as a generic failure,
erasing the distinction between "too slow for its window" and "broken".

A narrower except after a broader one is dead code, and no linter is configured
here to say so.

One trap in fixing it: while the branch was unreachable a timeout travelled the
normal path, which DOES update scheduled_tasks. Making the branch reachable
without that write would have traded a wrong status for a stale one, so
_update_task_outcome now mirrors terminal outcomes onto the parent row.

The timeout message also states that the work may still be running -- after
T-74 executors are handed to asyncio.to_thread, and a thread cannot be
cancelled, so wait_for frees the loop while the work continues.

Both fixes are mutation-checked: removing the startup call fails the ordering
test, removing the narrow except clause fails the propagation test. Suite goes
118 -> 126 passing with the same 36 pre-existing failures.

Co-Authored-By: Claude <noreply@anthropic.com>

src/main.py

@@ -65,6 +65,13 @@ async def lifespan(app: FastAPI):
     task_executor = TaskExecutor(settings)
     logger.info("Task executor initialized (max 5 concurrent tasks)")
 
+    # Before the first minute is processed, release any task still held by an
+    # execution row belonging to an instance that no longer exists. A 'running'
+    # row excludes its task from scheduling permanently, so skipping this leaves
+    # tasks silently unschedulable across every restart. Blocking briefly is fine
+    # here — the app serves no requests until lifespan yields.
+    task_executor.reconcile_orphaned_executions()
+
     # Initialize APScheduler with minimal configuration
     # No jobstore needed - we only have one in-memory job
     scheduler = AsyncIOScheduler(

src/tasks/executor.py

@@ -115,6 +115,70 @@ class TaskExecutor:
 
         return True
 
+    def reconcile_orphaned_executions(self) -> int:
+        """Close out execution rows left 'running' by a process that is gone.
+
+        get_tasks_for_minute excludes any task holding a 'running' row. That row
+        is written before the executor runs and updated after, so a process that
+        dies in between leaves it 'running' forever — and the task is then
+        excluded from every future minute, permanently, with no error and no log
+        line. It does not fail; it goes quiet, and quiet reads as healthy.
+
+        Live exposure rather than theory: Watchtower restarts this container at
+        4 AM daily, and the config backup starts at 03:05 and runs ~21 minutes.
+        A row from 2025-12-07 sat 'running' for eight months before anyone
+        looked.
+
+        Called at startup, where the reasoning is sound by construction: this
+        process has just begun, so nothing it can see is genuinely running, and
+        any such row belongs to an instance that no longer exists.
+
+        Marked 'orphaned', not 'failed'. When the process dies mid-task the work
+        may well have finished — a backup that completed and never got to update
+        its row is indistinguishable from one that died halfway. 'failed' would
+        assert an outcome nobody observed. 'orphaned' says only what is known:
+        we lost track of it.
+
+        Deliberately not extended to a time-based sweep of long-running rows.
+        While this process lives, execute_task's finally clause always closes the
+        row out, so a stale row implies a dead owner. A duration-based rule would
+        have to tell a slow task from a dead one, and getting that wrong closes
+        the record of a task that is still working.
+        """
+        try:
+            with self.get_db_connection() as conn:
+                with conn.cursor() as cur:
+                    cur.execute("""
+                        UPDATE task_executions
+                        SET status = 'orphaned',
+                            completed_at = %s,
+                            error = 'Scheduler restarted while this execution was '
+                                    'running; its outcome is unknown.'
+                        WHERE status = 'running'
+                        RETURNING task_name, started_at
+                    """, (datetime.now(timezone.utc),))
+                    orphans = cur.fetchall()
+                    conn.commit()
+        except Exception as e:
+            # Never fatal. A scheduler that refuses to start because it could not
+            # tidy up is worse than one carrying a stale row.
+            logger.error(f"Could not reconcile orphaned executions: {e}")
+            return 0
+
+        for task_name, started_at in orphans:
+            logger.warning(
+                f"Orphaned execution recovered: {task_name} was left 'running' "
+                f"since {started_at}. That task had been excluded from scheduling "
+                f"until now."
+            )
+        if orphans:
+            logger.warning(
+                f"{len(orphans)} task(s) were unschedulable and are now released."
+            )
+        else:
+            logger.info("No orphaned executions to reconcile")
+        return len(orphans)
+
     async def execute_task(self, task: Dict[str, Any]):
         """
         Execute a single task with timeout and error handling.
@@ -178,8 +242,17 @@ class TaskExecutor:
 
         except asyncio.TimeoutError:
             logger.error(f"Task {task_name} timed out after {timeout}s")
-            self._update_execution_status(execution_id, 'timeout',
-                                         error=f"Task exceeded timeout of {timeout}s")
+            self._update_execution_status(
+                execution_id, 'timeout',
+                error=f"Task exceeded timeout of {timeout}s. The underlying work may "
+                      f"still be running — executors that use asyncio.to_thread hand "
+                      f"the work to a thread, and a thread cannot be cancelled.")
+            # scheduled_tasks has to be written here as well. On the normal path
+            # it is updated alongside the execution row, and while this branch was
+            # unreachable a timeout travelled that path as a 'failed' result — so
+            # last_status did stay current. Making the branch reachable without
+            # this call would swap one wrong status for a stale one.
+            self._update_task_outcome(task_id, 'timeout', started_at)
 
         except Exception as e:
             logger.error(f"Task {task_name} failed with exception: {e}")
@@ -214,11 +287,52 @@ class TaskExecutor:
 
             return result, None
 
+        except asyncio.TimeoutError:
+            # This clause must precede `except Exception`, and that ordering is
+            # the entire bug it fixes. Since Python 3.11 asyncio.TimeoutError IS
+            # the builtin TimeoutError, which inherits OSError -> Exception, so
+            # the broad handler below used to catch it first and convert it into
+            # an ordinary (None, error) tuple. execute_task then filed it as a
+            # generic 'failed', and its own `except asyncio.TimeoutError` branch
+            # was unreachable: zero 'timeout' rows across 18,785 executions and
+            # eight months of history.
+            #
+            # Re-raised rather than returned, because the distinction is the
+            # point: "too slow for its window" and "broken" call for different
+            # responses and were indistinguishable in the record.
+            raise
+
         except ModuleNotFoundError:
             return None, f"Executor module not found: {executor_name}"
         except Exception as e:
             return None, f"Executor error: {str(e)}\n{traceback.format_exc()}"
 
+    def _update_task_outcome(self, task_id: int, status: str, started_at: datetime):
+        """Mirror a terminal outcome onto scheduled_tasks.
+
+        The happy path writes task_executions and scheduled_tasks in one
+        transaction. The error branches historically wrote only the former, which
+        did not show while every timeout was being funnelled through the happy
+        path as a 'failed'. Once a branch bypasses that path it has to keep
+        last_run/last_status current itself, or the task list quietly reports the
+        previous run's outcome as though it were the latest.
+        """
+        try:
+            completed_at = datetime.now(timezone.utc)
+            with self.get_db_connection() as conn:
+                with conn.cursor() as cur:
+                    cur.execute("""
+                        UPDATE scheduled_tasks
+                        SET last_run = %s, last_status = %s,
+                            last_duration_seconds = %s, updated_at = %s
+                        WHERE id = %s
+                    """, (completed_at, status,
+                          int((completed_at - started_at).total_seconds()),
+                          completed_at, task_id))
+                    conn.commit()
+        except Exception as e:
+            logger.error(f"Failed to update task outcome for {task_id}: {e}")
+
     def _update_execution_status(self, execution_id: int, status: str, error: str = None):
         """Update execution record with final status."""
         if execution_id is None: