diff --git a/cecli/args.py b/cecli/args.py
index 96fb0619ed1..5ee878775f1 100644
--- a/cecli/args.py
+++ b/cecli/args.py
@@ -440,13 +440,9 @@ def get_parser(default_config_files, git_root):
     )
     group.add_argument(
         "--scheduler",
-        action=argparse.BooleanOptionalAction,
+        metavar="SCHEDULER_CONFIG_JSON",
+        help="Specify scheduler configuration as a JSON string",
         default=None,
-        help=(
-            "Force the workflow scheduler on (--scheduler) or off (--no-scheduler) for"
-            " this session. (default: auto, on when any enabled scheduled workflow"
-            " exists)"
-        ),
     )
     group.add_argument(
         "--system-one",
diff --git a/cecli/coders/base_coder.py b/cecli/coders/base_coder.py
index 43c637224d4..5372fd2f98a 100755
--- a/cecli/coders/base_coder.py
+++ b/cecli/coders/base_coder.py
@@ -2253,6 +2253,31 @@ async def compact_context_if_needed(self, force=False, message=""):
                 )
                 return
 
+        # Compaction can only shrink the conversation buckets counted above.
+        # If the fixed context (files added to the chat, repo map, system
+        # prompt, rules, tool schemas) already meets the limit on its own,
+        # summarizing cannot bring the context under the limit. Bail out with
+        # guidance instead of churning every turn, which is what a large file
+        # added to the chat otherwise causes.
+        non_compactable_tokens = all_tokens - compactable_tokens
+
+        if (
+            self.context_compaction_max_tokens
+            and non_compactable_tokens >= self.context_compaction_max_tokens
+        ):
+            if getattr(self, "_compaction_noop_warning_time", 0) + 300 < time.time():
+                self._compaction_noop_warning_time = time.time()
+                self.io.tool_warning(
+                    "Compaction cannot reduce the context: the fixed context (files in the"
+                    f" chat, repo map, prompts) is already ~{non_compactable_tokens:,}"
+                    f" tokens, at or above the ~{self.context_compaction_max_tokens:,} token"
+                    " compaction limit."
+                )
+                self.io.tool_output("- Use /drop to remove unneeded files from the chat")
+                self.io.tool_output("- Use /clear to clear the chat history")
+
+            return
+
         message_tokens = done_tokens + cur_tokens
         file_tokens = diff_tokens + file_context_tokens
         combined_tokens = done_tokens + cur_tokens + diff_tokens + file_context_tokens
@@ -2692,7 +2717,7 @@ def warm_cache_worker():
     async def check_tokens(self, messages):
         """Check if the messages will fit within the model's token limits."""
         input_tokens = self.get_active_model().token_count(messages)
-        max_input_tokens = self.get_active_model().info.get("max_input_tokens") or 0
+        max_input_tokens = self.get_active_model().info.get("max_input_tokens") or 128000
 
         if max_input_tokens and input_tokens >= max_input_tokens:
             if (
@@ -2703,7 +2728,7 @@ async def check_tokens(self, messages):
                     f"Estimated chat context of {input_tokens:,} tokens exceeds the"
                     f" {max_input_tokens:,} token limit. Attempting to compact..."
                 )
-                await self.compact_context_if_needed(force=True)
+                await self.compact_context_if_needed()
 
                 # After compaction, re-format messages and re-check tokens
                 messages = self.format_messages()
@@ -3996,7 +4021,7 @@ async def show_send_output(self, completion):
         if (
             not len(self.partial_response_content)
             and not len(self.partial_response_tool_calls)
-            and not len(self.partial_response_reasoning_content)
+            and not _is_meaningful_reasoning(self.partial_response_reasoning_content)
         ):
             self.empty_response = True
             return
@@ -4126,7 +4151,8 @@ async def show_send_output_stream(self, completion):
 
                             text += reasoning_content
                             self.got_reasoning_content = True
-                            received_content = True
+                            if _is_meaningful_reasoning(reasoning_content):
+                                received_content = True
 
                         self.token_profiler.on_token()
                         self.io.update_spinner_suffix(reasoning_content)
@@ -4202,7 +4228,14 @@ async def show_send_output_stream(self, completion):
             self.io.tool_warning("Execution stopped by on message hook")
             return
 
-        if not received_content and len(self.partial_response_tool_calls) == 0:
+        # Treat the response as empty when nothing was received, or when the
+        # only thing received was reasoning made entirely of non-alphanumeric
+        # characters (e.g. moonshotai/kimi-k3 returning "!!!!").
+        if (
+            not received_content
+            and len(self.partial_response_tool_calls) == 0
+            and not _is_meaningful_reasoning(self.partial_response_reasoning_content)
+        ):
             self.empty_response = True
             return
 
@@ -5391,3 +5424,15 @@ def _first_usage_tokens(usage: object, paths: list[str], default: int = 0) -> in
         if value is not None:
             return value
     return default
+
+
+def _is_meaningful_reasoning(text):
+    """Return True if reasoning text contains at least one alphanumeric character.
+
+    Some providers (e.g. moonshotai/kimi-k3) occasionally return completions
+    with empty ``content`` and a ``reasoning_content`` made entirely of
+    punctuation (e.g. ``"!!!!"``). Those responses are effectively empty, so
+    the empty-response detector only lets reasoning count as response
+    content when it holds at least one alphanumeric character.
+    """
+    return bool(text) and any(ch.isalnum() for ch in text)
diff --git a/cecli/helpers/agents/service.py b/cecli/helpers/agents/service.py
index 6e2e28d8f1a..0dae1b08619 100644
--- a/cecli/helpers/agents/service.py
+++ b/cecli/helpers/agents/service.py
@@ -6,6 +6,7 @@
 
 import asyncio
 import logging
+import threading
 import time
 import weakref
 from dataclasses import dataclass
@@ -80,8 +81,22 @@ class AgentService:
     _uuid_coder_map: Dict[str, weakref.ref] = {}
     # Lock pools keyed by parent UUID — created lazily so only parents that
     # actually use them allocate a lock.
-    _spawn_locks: Dict[str, asyncio.Lock] = {}
-    _conversation_locks: Dict[str, asyncio.Lock] = {}
+    #
+    # Plain ``threading.Lock`` (not ``asyncio.Lock``): an ``asyncio.Lock`` binds
+    # itself to the event loop of its first *waiting* acquirer and raises
+    # ``RuntimeError('... is bound to a different event loop')`` when awaited
+    # from another loop. Spawn, reap and conversation writes all reach these
+    # pools from both the TUI main (Textual) loop and the coder worker-thread
+    # loop, and the pools are class-level, so a lock created before a hot reload
+    # outlives the loop it bound to. Every guarded section is fully synchronous,
+    # so a thread lock gives true cross-loop mutual exclusion with no loop
+    # affinity and is immune to task cancellation.
+    #
+    # Invariant: never hold one of these locks across an ``await`` — the guarded
+    # sections must stay synchronous, or the blocking acquire could stall the
+    # event loop.
+    _spawn_locks: Dict[str, threading.Lock] = {}
+    _conversation_locks: Dict[str, threading.Lock] = {}
     _primary_agent_uuid: str = None
 
     # ------------------------------------------------------------------ #
@@ -305,10 +320,15 @@ def max_sub_agents(self) -> int:
     # ------------------------------------------------------------------ #
     # Internal helpers
     @classmethod
-    def _get_lock(cls, pool: Dict[str, asyncio.Lock], uuid: str) -> asyncio.Lock:
-        """Return a lock for *uuid* from *pool*, creating one if absent."""
+    def _get_lock(cls, pool: Dict[str, threading.Lock], uuid: str) -> threading.Lock:
+        """Return a lock for *uuid* from *pool*, creating one if absent.
+
+        Locks are ``threading.Lock`` rather than ``asyncio.Lock`` (see the pool
+        declarations): the guarded sections are synchronous and correct locking
+        must work from any event loop.
+        """
         if uuid not in pool:
-            pool[uuid] = asyncio.Lock()
+            pool[uuid] = threading.Lock()
         return pool[uuid]
 
     @staticmethod
@@ -444,7 +464,7 @@ def _has_unfinished_descendants(agent_uuid: str) -> bool:
         # Acquire the spawn lock for the primary coder to serialise with
         # concurrent spawn operations that also hold this lock.
         parent_coder = parent if parent is not None else self.coder
-        async with self._get_lock(self._spawn_locks, parent_coder.uuid):
+        with self._get_lock(self._spawn_locks, parent_coder.uuid):
             for coder_uuid in list(self._sub_agent_order):
                 info = self.sub_agents.get(coder_uuid)
                 if (
@@ -593,7 +613,7 @@ async def _create_sub_agent_coder(
         # tried to acquire the same lock).
         parent_coder = parent if parent is not None else self.coder
 
-        async with self._get_lock(self._spawn_locks, parent_coder.uuid):
+        with self._get_lock(self._spawn_locks, parent_coder.uuid):
             self._check_max_sub_agents()
             new_uuid = str(uuid4()).split("-")[0]
 
@@ -686,7 +706,7 @@ async def _create_sub_agent_coder(
 
         # Re-acquire the lock to register — we must re-check max agents since
         # the lock was released and other spawns may have registered in between.
-        async with self._get_lock(self._spawn_locks, parent_coder.uuid):
+        with self._get_lock(self._spawn_locks, parent_coder.uuid):
             self._check_max_sub_agents()
 
             # Register in global coder lookup
@@ -952,7 +972,7 @@ async def _inject_sub_agent_result(self, info: SubAgentInfo) -> None:
         else:
             return
 
-        async with self._get_lock(self._conversation_locks, info.parent_uuid):
+        with self._get_lock(self._conversation_locks, info.parent_uuid):
             ConversationService.get_manager(parent_coder).add_message(
                 message_dict={"role": "user", "content": user_content},
                 tag=MessageTag.CUR,
@@ -1200,8 +1220,11 @@ def _tui_notify(tui, callback, *args):
     """Invoke *callback* on the TUI app, from either thread.
 
     ``call_from_thread`` only works when the caller is not the app's own
-    thread; scheduler dispatch runs on the primary (app) event loop, so fall
-    back to calling directly, which is correct when already on that thread.
+    thread. Scheduler-dispatched workflow runs now execute on the coder's own
+    loop (the TUI worker-thread loop — see ``SchedulerService._on_coder_loop``),
+    so notifications originate from the worker thread and ``call_from_thread``
+    applies; when we are already on the app thread (or the TUI is unavailable),
+    fall back to calling directly.
     """
 
     try:
diff --git a/cecli/helpers/io_proxy.py b/cecli/helpers/io_proxy.py
index 0dfb87154b4..f6963082102 100644
--- a/cecli/helpers/io_proxy.py
+++ b/cecli/helpers/io_proxy.py
@@ -183,7 +183,28 @@ async def get_input(self, *args, **kwargs):
 
     async def confirm_ask(self, *args, **kwargs):
         """Forward confirm_ask — per-coder queue iteration is handled by
-        TextualInputOutput which now iterates all per-coder queues."""
+        TextualInputOutput which now iterates all per-coder queues.
+
+        Coders configured with ``skip_cli_confirmations`` (unattended ``wf:``
+        workflow and ``ws:`` workspace workers) are auto-confirmed. They run
+        with nobody watching, and — more importantly — a blocking prompt has
+        to be awaited from whatever loop the coder happens to run on. A
+        scheduler-dispatched workflow runs on the main (Textual) loop, so
+        blocking there used to rebind ``queues``' global input-wake slot and
+        strand the primary coder's input pump.
+        """
+        coder_ref = getattr(self, "_coder", None)
+        coder = coder_ref() if coder_ref is not None else None
+
+        if coder is not None and getattr(coder, "skip_cli_confirmations", False):
+            question = args[0] if args else kwargs.get("question", "")
+            logger.debug(
+                "auto-confirming %r for unattended coder %s (skip_cli_confirmations)",
+                question,
+                self._coder_uuid,
+            )
+            return True
+
         if "coder_uuid" not in kwargs:
             kwargs["coder_uuid"] = self._coder_uuid
         return await self._target.confirm_ask(*args, **kwargs)
diff --git a/cecli/helpers/llms/domains/bedrock.py b/cecli/helpers/llms/domains/bedrock.py
index 68681504bab..ba6f7f3d809 100644
--- a/cecli/helpers/llms/domains/bedrock.py
+++ b/cecli/helpers/llms/domains/bedrock.py
@@ -194,7 +194,7 @@ def bedrock_endpoint(resolved: Dict[str, Any]) -> str:
         )
 
     route = resolved["route"]
-    return f"https://bedrock-runtime.{region}.amazonaws.com/model/{route}/converse"
+    return f"{resolved['api_base']}/model/{route}/converse"
 
 
 def _signed_headers(
diff --git a/cecli/helpers/llms/litellm_compat.py b/cecli/helpers/llms/litellm_compat.py
index 4086f141ab0..cb6c4b638a3 100644
--- a/cecli/helpers/llms/litellm_compat.py
+++ b/cecli/helpers/llms/litellm_compat.py
@@ -19,6 +19,7 @@
 
 import asyncio
 import json
+import math
 import os
 import re
 import warnings
@@ -1096,15 +1097,11 @@ def encode(
         """Estimate tokens for ``text`` without loading a BPE tokenizer.
 
         tiktoken's cl100k_base/o200k_base encodings cost ~30-95MB of RSS the
-        first time they are loaded, so we approximate instead: roughly one token
-        per four ASCII characters (the classic rule of thumb), plus one token per
-        non-ASCII character (CJK/emoji tokenize at about one token per char), and
-        never fewer tokens than there are whitespace-delimited word/punctuation
-        runs (dense code with short tokens is otherwise undercounted).
-
-        The estimate lands within ~±30% of cl100k_base on code and prose, which
-        is plenty for context-window checks, cost display, and file-size
-        warnings.
+        first time they are loaded, so we approximate instead with a regex
+        pre-tokenizer patterned on OpenAI's o200k_base (see
+        ``_estimate_token_count``). The estimate lands within ~±30% of
+        cl100k_base on code and prose, which is plenty for context-window
+        checks, cost display, and file-size warnings.
 
         Returns a ``range`` whose length is the estimated token count; the
         individual values are dummy ids (callers only use ``len()``/iteration).
@@ -1244,23 +1241,41 @@ def _load_litellm(self) -> None:
 # Token-count estimator (tiktoken-free)
 # ---------------------------------------------------------------------------
 
-_WORD_OR_PUNCT_RE = re.compile(r"\w+|[^\w\s]")
-_NON_ASCII_RE = re.compile(r"[^\x00-\x7f]")
+# Standard ``re`` approximation of OpenAI's o200k_base pre-tokenizer pattern.
+# The letter class is Unicode-aware (``[^\W\d_]``) so non-Latin scripts are
+# counted rather than dropped.
+_O200K_APPROX_PATTERN = re.compile(
+    r"""'(?:[sdmt]|ll|ve|re)| ?[^\W\d_]+| ?\d{1,3}| ?[^\s\w]+|\s+""", re.UNICODE
+)
+
+#: Bytes charged per token for chunks too long to be a single token, calibrated
+#: against o200k_base (~5 bytes/token on code, prose and JSON).
+_CHARS_PER_TOKEN = 5.0
 
 
 def _estimate_token_count(text: str) -> int:
     """Estimate the number of tokens in ``text`` without a BPE tokenizer.
 
-    Uses the classic ~4 chars/token rule of thumb, lifted for non-ASCII text
-    (CJK/emoji tokenize at roughly one token per character) and floored by the
-    number of word/punctuation runs so dense code with many short tokens isn't
-    undercounted.
+    Splits ``text`` with a regex approximation of OpenAI's o200k_base
+    pre-tokenizer, then accumulates a fractional token weight per chunk:
+    whitespace runs collapse to a single token, numbers and short chunks
+    (<= 4 bytes) count as one, and longer chunks are charged at roughly
+    ``_CHARS_PER_TOKEN`` bytes per token. The total is rounded up and floored
+    at one token for any non-empty input.
     """
 
     if not text:
         return 0
 
-    units = len(_WORD_OR_PUNCT_RE.findall(text))
-    non_ascii = len(_NON_ASCII_RE.findall(text))
+    tokens = 0.0
+    for chunk in _O200K_APPROX_PATTERN.findall(text):
+        byte_len = len(chunk.encode("utf-8"))
+
+        if chunk.isspace():
+            tokens += 1.0
+        elif chunk.strip().isdigit() or byte_len <= 4:
+            tokens += 1.0
+        else:
+            tokens += max(1.0, byte_len / _CHARS_PER_TOKEN)
 
-    return max(1, len(text) // 4, units) + non_ascii
+    return max(1, math.ceil(tokens - 1e-9))
diff --git a/cecli/helpers/queues.py b/cecli/helpers/queues.py
index 3f868d97812..2ccd59e54f6 100644
--- a/cecli/helpers/queues.py
+++ b/cecli/helpers/queues.py
@@ -3,9 +3,12 @@
 from __future__ import annotations
 
 import asyncio
+import logging
 import queue
 from typing import Any
 
+logger = logging.getLogger(__name__)
+
 # Global registry: coder_uuid → queue.Queue for sending input to that coder
 _per_coder_queues: dict[str, "queue.Queue"] = {}
 
@@ -19,6 +22,13 @@
 _input_loop: asyncio.AbstractEventLoop | None = None
 _input_wake: asyncio.Event | None = None
 
+# Dedicated wake events for consumers parked on a loop *other* than the primary
+# input loop (for example a scheduler-dispatched workflow asking for input while
+# the scheduler ticks on the main/Textual loop). They get their own slot so a
+# foreign-loop consumer can never steal the primary binding and strand the
+# primary coder's input pump.
+_input_wakes: dict[asyncio.AbstractEventLoop, asyncio.Event] = {}
+
 
 def register_coder_queue(coder_uuid: str, q: "queue.Queue") -> None:
     """Register a per-coder input queue.
@@ -83,6 +93,9 @@ def set_input_loop(loop: asyncio.AbstractEventLoop) -> None:
     global _input_loop, _input_wake
     _input_loop = loop
     _input_wake = asyncio.Event()
+    # This loop now owns the primary slot, so drop any dedicated wake it may
+    # have registered while it was still a foreign consumer.
+    _input_wakes.pop(loop, None)
 
 
 def wake_input_waiters() -> None:
@@ -95,39 +108,73 @@ def wake_input_waiters() -> None:
     coder worker loop), the stale binding is dropped so the next
     wait_for_input() rebinds to the current loop instead of raising
     "Event loop is closed".
+
+    Every loop with a parked consumer is woken, not just the primary one:
+    a consumer on a foreign loop has its own event (see ``_input_wakes``) so
+    the primary binding is never stolen from it.
     """
     global _input_loop, _input_wake
 
     loop = _input_loop
-    if loop is None or _input_wake is None:
-        return
 
-    if loop.is_closed():
-        _input_loop = None
-        _input_wake = None
-        return
+    if loop is not None:
+        if _input_wake is None:
+            _input_loop = None
+        elif loop.is_closed():
+            _input_loop = None
+            _input_wake = None
+        else:
+            loop.call_soon_threadsafe(_input_wake.set)
+
+    for other, wake in list(_input_wakes.items()):
+        if other.is_closed():
+            _input_wakes.pop(other, None)
+            continue
 
-    loop.call_soon_threadsafe(_input_wake.set)
+        try:
+            other.call_soon_threadsafe(wake.set)
+        except RuntimeError:
+            # Loop stopped accepting callbacks between the check and the call.
+            _input_wakes.pop(other, None)
 
 
 async def wait_for_input() -> None:
     """Await the next input push without polling.
 
-    Must be called from the input loop (the coder worker loop). Consumers
-    sweep the payload queues first, then block here until wake_input_waiters()
-    fires. Initializes the wake state from the running loop on first use and
-    rebinds whenever the previous loop was closed or is no longer the
-    running loop (which happens across a hot reload).
+    Intended to be called from the input loop (the coder worker loop).
+    Consumers sweep the payload queues first, then block here until
+    wake_input_waiters() fires. Initializes the wake state from the running
+    loop on first use and rebinds whenever the previous loop was closed (which
+    happens across a hot reload).
+
+    A caller on a *different*, still-live loop no longer rebinds the primary
+    slot: it gets a dedicated wake event instead. Rebinding used to strand the
+    primary coder's input pump on the orphaned event, which froze the TUI
+    (prompts, slash commands and ``/exit`` all unresponsive; only Ctrl-Q
+    worked) whenever a scheduler-dispatched workflow needed input or a
+    confirmation and therefore ran on the main (Textual) loop.
     """
     loop = asyncio.get_running_loop()
 
-    if (
-        _input_loop is None
-        or _input_wake is None
-        or _input_loop.is_closed()
-        or _input_loop is not loop
-    ):
+    if _input_loop is None or _input_wake is None or _input_loop.is_closed():
         set_input_loop(loop)
 
-    _input_wake.clear()
-    await _input_wake.wait()
+    if _input_loop is loop:
+        _input_wake.clear()
+        await _input_wake.wait()
+        return
+
+    logger.warning(
+        "wait_for_input() called from a non-input loop (%r); using a dedicated"
+        " wake event instead of rebinding the primary input loop",
+        loop,
+    )
+
+    wake = _input_wakes.get(loop)
+
+    if wake is None:
+        wake = asyncio.Event()
+        _input_wakes[loop] = wake
+
+    wake.clear()
+    await wake.wait()
diff --git a/cecli/helpers/scheduler/service.py b/cecli/helpers/scheduler/service.py
index a6d1ad493ca..f3d81c1a8e8 100644
--- a/cecli/helpers/scheduler/service.py
+++ b/cecli/helpers/scheduler/service.py
@@ -173,7 +173,9 @@ async def _dispatch(self, wf: Any, lease: Any, slot: Any, now: datetime) -> None
             renewal = asyncio.create_task(renewal)
 
         try:
-            result = await execute_workflow(self.coder, wf, source="scheduler")
+            result = await self._on_coder_loop(
+                lambda: execute_workflow(self.coder, wf, source="scheduler")
+            )
             await self.db.finish(lease, result)
             await self._log_tasks(lease, result)
         except asyncio.CancelledError:
@@ -191,6 +193,43 @@ async def _dispatch(self, wf: Any, lease: Any, slot: Any, now: datetime) -> None
             nxt = self._successor(wf, slot, now, next_run)
             await self.db.advance(wf.name, nxt)
 
+    def _coder_loop(self) -> Optional[asyncio.AbstractEventLoop]:
+        """Best-effort discovery of the loop the interactive coder runs on.
+
+        In TUI mode the coder lives on a worker-thread loop while the scheduler
+        ticks on the main (Textual) loop, so the two differ. Returns ``None``
+        when the coder's loop cannot be determined (linear/one-shot sessions,
+        tests), in which case callers stay on the current loop.
+        """
+
+        tui_ref = getattr(self.coder, "tui", None)
+
+        try:
+            tui = tui_ref() if callable(tui_ref) else tui_ref
+        except Exception:
+            tui = None
+
+        worker = getattr(tui, "worker", None)
+        loop = getattr(worker, "loop", None)
+
+        return loop if isinstance(loop, asyncio.AbstractEventLoop) else None
+
+    async def _on_coder_loop(self, make_coro: Any) -> Any:
+        """Await ``make_coro()`` on the coder's own event loop.
+
+        A workflow run owns coder I/O (prompts, confirmations, tool output),
+        which is bound to the loop the coder actually runs on.
+        """
+
+        target = self._coder_loop()
+        running = asyncio.get_running_loop()
+
+        if target is None or target is running or target.is_closed():
+            return await make_coro()
+
+        future = asyncio.run_coroutine_threadsafe(make_coro(), target)
+        return await asyncio.wrap_future(future)
+
     async def _log_tasks(self, lease: Any, result: Any) -> None:
         """Mirror the runner's per-task outcomes into the DB ``task_log``."""
 
diff --git a/cecli/helpers/workflows/config.py b/cecli/helpers/workflows/config.py
index 3b142767c7a..8b49b955744 100644
--- a/cecli/helpers/workflows/config.py
+++ b/cecli/helpers/workflows/config.py
@@ -10,8 +10,10 @@
    ``None``, ``''``, ``'{}'``, ``'[]'`` and ``'null'`` count as *absent* so that
    lower-precedence sources still load.
 2. ``<git_root or cwd>/.cecli.workflows.yml`` / ``.cecli.workflows.yaml``.
-3. ``merged_config["workflows"]`` — the deep-merged ``.cecli.conf.yml`` value.
-4. ``~/.cecli/workflows/*.yml`` / ``*.yaml`` — globbed in stable sorted order.
+3. ``<git_root or cwd>/.cecli/workflows/*.yml`` / ``*.yaml`` — globbed in stable sorted order.
+4. ``<git_root or cwd>/.agents/workflows/*.yml`` / ``*.yaml`` — globbed in stable sorted order.
+5. ``merged_config["workflows"]`` — the deep-merged ``.cecli.conf.yml`` value.
+6. ``~/.cecli/workflows/*.yml`` / ``*.yaml`` — globbed in stable sorted order.
 
 Each source may hold a single workflow mapping or a ``workflows: [...]`` list.
 On a duplicate ``name`` the higher-precedence definition wins and the loser is
@@ -41,6 +43,8 @@
 
 WORKFLOW_DIR = Path(".cecli") / "workflows"
 
+AGENTS_WORKFLOW_DIR = Path(".agents") / "workflows"
+
 # Values argparse/deep-merge normalization leaves behind when nothing was set.
 _ABSENT_STRINGS = ("", "{}", "[]", "null", "~")
 
@@ -93,6 +97,8 @@ def resolve_workflow_configs(
         ordered += _entries_from_arg(str(config_arg))
 
     ordered += _entries_from_local(git_root)
+    ordered += _entries_from_local_dir(git_root)
+    ordered += _entries_from_local_agents_dir(git_root)
 
     if isinstance(merged_config, dict):
         ordered += _entries(merged_config.get("workflows"), None, _CONFIG_LABEL)
@@ -289,9 +295,28 @@ def _entries_from_local(git_root: Optional[Any]) -> List[_Discovered]:
     return []
 
 
+def _entries_from_local_dir(git_root: Optional[Any]) -> List[_Discovered]:
+    """Resolve a repo-local ``.cecli/workflows/*.yml|*.yaml`` directory."""
+    base = Path(git_root).expanduser() if git_root else Path.cwd()
+
+    return _entries_from_dir(base / WORKFLOW_DIR)
+
+
+def _entries_from_local_agents_dir(git_root: Optional[Any]) -> List[_Discovered]:
+    """Resolve a repo-local ``.agents/workflows/*.yml|*.yaml`` directory."""
+    base = Path(git_root).expanduser() if git_root else Path.cwd()
+
+    return _entries_from_dir(base / AGENTS_WORKFLOW_DIR)
+
+
 def _entries_from_home() -> List[_Discovered]:
     """Resolve ``~/.cecli/workflows/*.yml|*.yaml`` in stable sorted order."""
-    root = Path.home() / WORKFLOW_DIR
+
+    return _entries_from_dir(Path.home() / WORKFLOW_DIR)
+
+
+def _entries_from_dir(root: Path) -> List[_Discovered]:
+    """Resolve ``*.yml`` / ``*.yaml`` files under *root* in stable sorted order."""
     if not root.is_dir():
         return []
 
diff --git a/cecli/main.py b/cecli/main.py
index 450ad919f34..4c93fbbd40e 100644
--- a/cecli/main.py
+++ b/cecli/main.py
@@ -173,6 +173,7 @@ def convert_yaml_to_json_string(value, config_file_value=None):
     "system_one": "system-one",
     "server_config": "server-config",
     "workflows": "workflows",
+    "scheduler": "scheduler",
 }
 
 
@@ -1212,6 +1213,16 @@ def get_io(pretty):
     lint_cmds = parse_lint_cmds(args.lint_cmd, io)
     if lint_cmds is None:
         return await graceful_exit(None, 1)
+
+    # Create/attach the git repo before the coder is built. The coder resolves
+    # its repo, root, and repo_map at construction; if the repo is created
+    # afterwards the coder keeps repo=None for the whole session and
+    # repo-backed tools (Grep, git_*, ...) fail with "Not in a git repository."
+    if args.git and not (suppress_pre_init or args.workspaces or args.workspace_name):
+        git_root = await setup_git(git_root, pre_init_io)
+        if args.gitignore:
+            await check_gitignore(git_root, pre_init_io)
+
     repo = None
     if args.git:
         try:
@@ -1382,10 +1393,6 @@ def get_io(pretty):
                     pre_init_io.tool_output()
                 except KeyboardInterrupt:
                     return await graceful_exit(coder, 1)
-        if args.git and not (suppress_pre_init or args.workspaces or args.workspace_name):
-            git_root = await setup_git(git_root, pre_init_io)
-            if args.gitignore:
-                await check_gitignore(git_root, pre_init_io)
     except UnknownEditFormat as err:
         pre_init_io.tool_error(str(err))
         await pre_init_io.offer_url(
@@ -1539,17 +1546,17 @@ def get_io(pretty):
     # --message/--exit paths above return before the scheduler can tick.
     # Auto mode turns it on iff at least one enabled scheduled workflow
     # exists, so nothing ever touches the coordination DB otherwise.
-    _scheduler_cfg = merged_config.get("scheduler") or {}
-    if _workflows and _scheduler_enabled(args, _scheduler_cfg, _workflows):
+    _scheduler_cfg = _scheduler_config(args)
+    if _workflows and _scheduler_enabled(_scheduler_cfg, _workflows):
         from cecli.helpers.scheduler import SchedulerService
 
         coder.scheduler = SchedulerService(
             coder,
             _workflows,
             io,
-            tick_seconds=_scheduler_cfg.get("tick_seconds", 60),
-            catch_up=_scheduler_cfg.get("catch_up", True),
-            max_catchup=_scheduler_cfg.get("max_catchup", 1),
+            tick_seconds=_scheduler_cfg["tick_seconds"],
+            catch_up=_scheduler_cfg["catch_up"],
+            max_catchup=_scheduler_cfg["max_catchup"],
         )
         await coder.scheduler.start()
 
@@ -1685,18 +1692,46 @@ def get_io(pretty):
             raise
 
 
-def _scheduler_enabled(args, scheduler_cfg, workflows):
-    """Resolve whether the scheduler should run: CLI flag > config > auto."""
+def _scheduler_enabled(scheduler_cfg, workflows):
+    """Resolve whether the scheduler should run: explicit ``enabled`` > auto."""
 
-    if getattr(args, "scheduler", None) is not None:
-        return bool(args.scheduler)
-
-    if scheduler_cfg.get("enabled", None) is not None:
-        return bool(scheduler_cfg.get("enabled"))
+    if scheduler_cfg.get("enabled") is not None:
+        return bool(scheduler_cfg["enabled"])
 
     return any(getattr(wf, "schedule", None) and getattr(wf, "enabled", True) for wf in workflows)
 
 
+def _scheduler_config(args):
+    """Resolve the ``scheduler`` config object, defaults layered underneath.
+
+    ``args.scheduler`` is a JSON string (the CLI ``--scheduler`` value deep-merged
+    over the config-file object) or ``None`` when unset, in which case the
+    documented defaults apply and ``enabled`` stays ``None`` (auto).
+    """
+
+    defaults = {
+        "enabled": None,
+        "tick_seconds": 60,
+        "catch_up": True,
+        "max_catchup": 1,
+    }
+
+    raw = getattr(args, "scheduler", None)
+
+    if not raw:
+        return dict(defaults)
+
+    try:
+        parsed = json.loads(raw)
+    except (json.JSONDecodeError, TypeError):
+        return dict(defaults)
+
+    if not isinstance(parsed, dict):
+        return dict(defaults)
+
+    return {**defaults, **parsed}
+
+
 async def _stop_scheduler_quietly(coder):
     """Best-effort scheduler shutdown used by reload/exit paths."""
 
diff --git a/cecli/watch.py b/cecli/watch.py
index f7418147666..4c9e1be094a 100644
--- a/cecli/watch.py
+++ b/cecli/watch.py
@@ -322,6 +322,12 @@ def catch_up_scan(self):
         # before this scan can share the threshold tick and must not be re-found.
         threshold = self.last_scan_time
         found = set()
+        # Track the newest filesystem mtime observed during this scan. On
+        # Windows the wall clock (time.time) is coarser than file mtimes, so a
+        # file written just before this scan can carry an mtime slightly ahead
+        # of scan_started; advancing the threshold from the filesystem clock
+        # instead of the wall clock keeps such a file from being re-found.
+        max_mtime = threshold
 
         for root in self.get_roots_to_watch():
             root_path = Path(root)
@@ -335,12 +341,15 @@ def catch_up_scan(self):
 
             for path in candidates:
                 try:
+                    max_mtime = max(max_mtime, path.stat().st_mtime)
                     if self.filter_func(None, str(path)):
                         found.add(str(path.absolute()))
                 except Exception:
                     continue
 
-        self.last_scan_time = scan_started
+        # Advance to the newest filesystem timestamp we saw so the next scan's
+        # threshold is never behind a file's mtime (see note above).
+        self.last_scan_time = max(scan_started, max_mtime)
 
         if found:
             self.changed_files.update(found)
diff --git a/cecli/website/docs/config/retries.md b/cecli/website/docs/config/retries.md
index 3140c2de6ed..0aa01c39f85 100644
--- a/cecli/website/docs/config/retries.md
+++ b/cecli/website/docs/config/retries.md
@@ -11,6 +11,7 @@ Cecli can be configured to retry failed API calls. This is useful for handling i
 - `retry-timeout`: The timeout in seconds for each retry.
 - `retry-backoff-factor`: The backoff factor to use between retries.
 - `retry-on-unavailable`: Whether to retry on 503 Service Unavailable errors.
+- `retry-on-empty`: Whether to retry when the model returns an empty response. A response is considered empty when it has no content, no tool calls, and no *meaningful* reasoning. Reasoning that contains no alphanumeric characters (for example a `reasoning_content` of `"!!!!"`) does not count as a response, so it is retried like any other empty response.
 - `retry-on-unauthorized`: Whether to retry on 401 Unauthorized (and 403 Forbidden) errors. Default: false.
 
 Example usage in `.cecli.conf.yml`:
diff --git a/cecli/website/docs/config/scheduler.md b/cecli/website/docs/config/scheduler.md
index 1d976a7bd81..2ab66ec410e 100644
--- a/cecli/website/docs/config/scheduler.md
+++ b/cecli/website/docs/config/scheduler.md
@@ -1,7 +1,7 @@
 ---
 parent: Configuration
 nav_order: 48
-description: The scheduler turns workflow schedule: entries into runs, coordinating any number of cecli instances through a shared SQLite database.
+description: The scheduler turns workflow schedule entries into runs, coordinating any number of cecli instances through a shared SQLite database.
 ---
 # Scheduler
 
@@ -9,14 +9,7 @@ The [scheduler](workflows.html) is the component that turns a [workflow](workflo
 
 ## Running it
 
-The scheduler starts automatically in any interactive cecli session that has at least one enabled, scheduled workflow. Control it explicitly per session with:
-
-```bash
-cecli --scheduler       # force on for this session
-cecli --no-scheduler    # force off (workflows are still usable via /workflows run)
-```
-
-or persistently in `.cecli.conf.yml`:
+The scheduler starts automatically in any interactive cecli session that has at least one enabled, scheduled workflow. It is configured as an object in `.cecli.conf.yml`:
 
 ```yaml
 scheduler:
@@ -26,7 +19,9 @@ scheduler:
   max_catchup: 1              # how many missed slots to replay before skipping ahead
 ```
 
-Resolution order for whether it runs: the CLI flag (`--scheduler` / `--no-scheduler`) wins, then `scheduler.enabled` from config, then auto — on when any enabled workflow has a `schedule:`, off otherwise. Nothing touches the coordination database when no scheduled workflow exists, and one-shot (`--message`) runs return before the scheduler can tick.
+`--scheduler <json>` overrides it for a single session with the same object, e.g. `cecli --scheduler '{"enabled": false}'`. Disabling is always expressed through the object's `enabled` key — there is no separate off switch — and workflows remain usable manually via `/workflows run`.
+
+Resolution order for whether it runs: `enabled` (from the CLI `--scheduler <json>` deep-merged over the `scheduler:` config key) decides on/off; when `enabled` is omitted the mode is auto — on when any enabled workflow has a `schedule:`, off otherwise. Nothing touches the coordination database when no scheduled workflow exists, and one-shot (`--message`) runs return before the scheduler can tick.
 
 The `catch_up` pair controls what happens after downtime (e.g. a sleeping laptop):
 
@@ -35,10 +30,6 @@ The `catch_up` pair controls what happens after downtime (e.g. a sleeping laptop
 
 An individual workflow may override the global setting with its own top-level `catch_up: true|false` key (see [Workflows](workflows.html)); unset means "follow the scheduler".
 
-A headless daemon mode (scheduling on a server with no interactive session) is planned but not yet available; today scheduled runs only fire while a cecli session is alive.
-
-In TUI mode the scheduler ticks and dispatches on the primary event loop while the interactive coder runs on the TUI worker-thread loop. Scheduled runs still get their own sub-agent pane: the container notifications handle being called from the app's own thread. Workflow dispatch itself is loop-safe (the DB is threaded, sub-agent spawning marshals through `AgentService`), but a run's worker conversation shares MCP connections across loops; prefer linear (non-TUI) sessions or a dedicated `--scheduler` instance for production schedules until a `run_coroutine_threadsafe` bridge lands.
-
 ## Multi-instance coordination
 
 You can safely run **many** cecli instances at once (same machine, or one shared `~/.cecli` on a network filesystem — see caveats). They cooperate through a global SQLite database:
@@ -68,5 +59,6 @@ Scheduler state is visible from any session with the [workflow commands](workflo
 
 ## Arguments
 
-- `--scheduler` / `--no-scheduler`: Force the workflow scheduler on or off for this session.
+- `--scheduler <json>`: Override scheduler configuration for this session (same object as the `scheduler:` config key).
 - `--workflows <json|yaml>`: Workflow definitions the scheduler will consider (see [Workflows](workflows.html)).
+- To disable the scheduler, pass `--scheduler '{"enabled": false}'` (or set `enabled: false` under the `scheduler:` config key).
diff --git a/cecli/website/docs/config/workflows.md b/cecli/website/docs/config/workflows.md
index 9f1915fe868..5b4ccabdb16 100644
--- a/cecli/website/docs/config/workflows.md
+++ b/cecli/website/docs/config/workflows.md
@@ -17,12 +17,14 @@ Preferred: one YAML file per workflow in `~/.cecli/workflows/`:
     morning-triage.yml
 ```
 
-Also supported: inline under a `workflows:` key in `.cecli.conf.yml`, or a repo-local `.cecli.workflows.yml`. Each source may hold a single workflow mapping, a `workflows: [...]` list, or a plain list of mappings. Resolution precedence (highest first):
+Also supported: a repo-local `.cecli/workflows/` or `.agents/workflows/` directory (one YAML file per workflow), inline under a `workflows:` key in `.cecli.conf.yml`, or a repo-local `.cecli.workflows.yml`. Each source may hold a single workflow mapping, a `workflows: [...]` list, or a plain list of mappings. Resolution precedence (highest first):
 
 1. **CLI argument** — an inline JSON/YAML string or a file path passed to `--workflows`.
 2. **Local file** — `.cecli.workflows.yml` / `.cecli.workflows.yaml` in the git root (or working directory).
-3. **Merged config** — the `workflows:` key in `.cecli.conf.yml`.
-4. **Global files** — `~/.cecli/workflows/*.yml` / `*.yaml`, loaded in sorted order.
+3. **Local directory** — `.cecli/workflows/*.yml` / `*.yaml` in the git root (or working directory), loaded in sorted order.
+4. **Standard local directory** — `.agents/workflows/*.yml` / `*.yaml` in the git root (or working directory), loaded in sorted order.
+5. **Merged config** — the `workflows:` key in `.cecli.conf.yml`.
+6. **Global files** — `~/.cecli/workflows/*.yml` / `*.yaml`, loaded in sorted order.
 
 On a duplicate `name`, the higher-precedence definition wins and the loser is skipped with a warning. A source that fails to parse is skipped with a warning as well — a malformed workflows file never crashes startup.
 
@@ -152,4 +154,4 @@ tasks:
 ## Arguments
 
 - `--workflows <json|yaml>`: Provide an inline workflows configuration or a path to a workflows file, overriding all file-based sources.
-- `--scheduler` / `--no-scheduler`: Force the [scheduler](scheduler.html) on or off for this session.
+- `--scheduler <json>`: Override [scheduler](scheduler.html) configuration for this session (same object as the `scheduler:` config key).
diff --git a/tests/basic/test_agent_config_merge.py b/tests/basic/test_agent_config_merge.py
index b691666eab5..680dbb066b4 100644
--- a/tests/basic/test_agent_config_merge.py
+++ b/tests/basic/test_agent_config_merge.py
@@ -113,6 +113,7 @@ def test_all_yaml_to_json_args_deep_merge_with_config_file():
         "system_one",
         "server_config",
         "workflows",
+        "scheduler",
     }
     assert all("_" not in key for key in YAML_TO_JSON_ARG_KEYS.values())
 
diff --git a/tests/basic/test_empty_response.py b/tests/basic/test_empty_response.py
new file mode 100644
index 00000000000..4b562fc7222
--- /dev/null
+++ b/tests/basic/test_empty_response.py
@@ -0,0 +1,223 @@
+"""Tests for empty-response detection with non-meaningful reasoning content.
+
+Regression coverage for the ``retry-on-empty`` fix: a completion with empty
+``content``, no tool calls, and a ``reasoning_content`` made entirely of
+non-alphanumeric characters (e.g. moonshotai/kimi-k3 returning ``"!!!!"``) must
+be classified as an empty response so the existing retry loop engages.
+"""
+
+from types import SimpleNamespace
+
+import pytest
+
+from cecli.coders.base_coder import Coder, _is_meaningful_reasoning
+from cecli.helpers.threading import ThreadSafeEvent
+from cecli.llm import litellm
+
+
+# --------------------------------------------------------------------------- #
+# Test doubles
+# --------------------------------------------------------------------------- #
+class _AlwaysSetEvent:
+    def is_set(self):
+        return True
+
+
+class _FakeIO:
+    def __init__(self):
+        self.confirmation_in_progress_event = _AlwaysSetEvent()
+        self.assistant_outputs = []
+
+    def tool_error(self, *a, **k):
+        pass
+
+    def tool_warning(self, *a, **k):
+        pass
+
+    def update_spinner_suffix(self, *a, **k):
+        pass
+
+    def reset_streaming_response(self):
+        pass
+
+    def stream_output(self, *a, **k):
+        pass
+
+    def ai_output(self, *a, **k):
+        pass
+
+    def tool_output(self, *a, **k):
+        pass
+
+    def assistant_output(self, *a, **k):
+        self.assistant_outputs.append(a)
+
+
+class _FakeTokenProfiler:
+    def start(self):
+        pass
+
+    def on_token(self):
+        pass
+
+    def on_error(self):
+        pass
+
+    def add_to_usage_report(self, *a, **k):
+        return a[0] if a else ""
+
+
+def _make_coder(stream=False):
+    coder = Coder.__new__(Coder)
+    coder.stream = stream
+    coder.verbose = False
+    coder.args = SimpleNamespace(debug=False, show_thinking=False)
+    coder.io = _FakeIO()
+    coder.interrupt_event = ThreadSafeEvent()
+    coder.pretty = False
+    coder.reasoning_tag_name = "THINKING"
+    coder.got_reasoning_content = False
+    coder.ended_reasoning_content = False
+    coder.empty_response = False
+    coder.tool_reflection = False
+    coder.partial_response_content = ""
+    coder.partial_response_reasoning_content = ""
+    coder.partial_response_chunks = []
+    coder.partial_response_tool_calls = []
+    coder.partial_response_function_call = dict()
+    coder.partial_response_consolidated = None
+    coder.multi_response_content = ""
+    coder.chat_completion_response_hashes = []
+    coder._streaming_buffer_length = 0
+    coder.token_profiler = _FakeTokenProfiler()
+    coder._output_loop_detected = False
+    coder._output_loop_message = ""
+    coder._has_empty_reflected = False
+    coder.edit_format = "code"
+    return coder
+
+
+def _tc(index, call_id, name, arguments):
+    return litellm.ChatCompletionMessageToolCall(
+        id=call_id,
+        function=litellm.Function(arguments=arguments or "", name=name),
+        type="function",
+        index=index,
+    )
+
+
+def _non_streaming_completion(content="", reasoning=None, tool_calls=None):
+    message = {"role": "assistant", "content": content}
+    if reasoning is not None:
+        message["reasoning_content"] = reasoning
+    if tool_calls is not None:
+        message["tool_calls"] = tool_calls
+    return litellm.ModelResponse(
+        id="test-completion",
+        created=0,
+        model="gpt-test",
+        object="chat.completion",
+        choices=[{"finish_reason": "stop", "index": 0, "message": message}],
+        usage={"completion_tokens": 1, "prompt_tokens": 1, "total_tokens": 2},
+    )
+
+
+def _reasoning_chunk(text):
+    delta = litellm.Delta(role="assistant", content=None, reasoning_content=text)
+    choice = litellm.StreamChoice(finish_reason=None, index=0, delta=delta)
+    return litellm.StreamChunk(
+        id="cmpl-test", created=1000, model="gpt-test", choices=[choice], usage=None
+    )
+
+
+async def _agen(chunks):
+    for chunk in chunks:
+        yield chunk
+
+
+# --------------------------------------------------------------------------- #
+# _is_meaningful_reasoning unit tests
+# --------------------------------------------------------------------------- #
+def test_is_meaningful_reasoning_empty_and_none():
+    assert _is_meaningful_reasoning("") is False
+    assert _is_meaningful_reasoning(None) is False
+
+
+def test_is_meaningful_reasoning_punctuation_only():
+    assert _is_meaningful_reasoning("!!!!") is False
+    assert _is_meaningful_reasoning("...?!,;:") is False
+
+
+def test_is_meaningful_reasoning_whitespace_only():
+    assert _is_meaningful_reasoning("   \n\t  ") is False
+
+
+def test_is_meaningful_reasoning_normal_text():
+    assert _is_meaningful_reasoning("Let me think about this") is True
+
+
+def test_is_meaningful_reasoning_mixed_punctuation_and_alnum():
+    assert _is_meaningful_reasoning("42?") is True
+    assert _is_meaningful_reasoning("!!!a!!!") is True
+
+
+# --------------------------------------------------------------------------- #
+# Non-streaming path (show_send_output)
+# --------------------------------------------------------------------------- #
+@pytest.mark.asyncio
+async def test_junk_reasoning_is_empty_non_streaming():
+    coder = _make_coder(stream=False)
+    await coder.show_send_output(_non_streaming_completion(content="", reasoning="!" * 40))
+    assert coder.empty_response is True
+
+
+@pytest.mark.asyncio
+async def test_meaningful_reasoning_is_not_empty_non_streaming():
+    coder = _make_coder(stream=False)
+    await coder.show_send_output(
+        _non_streaming_completion(content="", reasoning="Let me think about this")
+    )
+    assert coder.empty_response is False
+
+
+@pytest.mark.asyncio
+async def test_tool_calls_are_not_empty_non_streaming():
+    coder = _make_coder(stream=False)
+    await coder.show_send_output(
+        _non_streaming_completion(
+            content="",
+            reasoning="",
+            tool_calls=[_tc(0, "call_1", "Local--ls", "{}")],
+        )
+    )
+    assert coder.empty_response is False
+
+
+@pytest.mark.asyncio
+async def test_mixed_reasoning_is_not_empty_non_streaming():
+    coder = _make_coder(stream=False)
+    await coder.show_send_output(_non_streaming_completion(content="", reasoning="42?"))
+    assert coder.empty_response is False
+
+
+# --------------------------------------------------------------------------- #
+# Streaming path (show_send_output_stream)
+# --------------------------------------------------------------------------- #
+@pytest.mark.asyncio
+async def test_junk_reasoning_is_empty_streaming():
+    coder = _make_coder(stream=True)
+    coder.args.show_thinking = True
+    async for _ in coder.show_send_output_stream(_agen([_reasoning_chunk("!" * 40)])):
+        pass
+    assert coder.empty_response is True
+
+
+@pytest.mark.asyncio
+async def test_meaningful_reasoning_is_not_empty_streaming():
+    coder = _make_coder(stream=True)
+    coder.args.show_thinking = True
+    async for _ in coder.show_send_output_stream(
+        _agen([_reasoning_chunk("Let me think about this")])
+    ):
+        pass
+    assert coder.empty_response is False
diff --git a/tests/basic/test_scheduler_config.py b/tests/basic/test_scheduler_config.py
new file mode 100644
index 00000000000..6b53bf3ef77
--- /dev/null
+++ b/tests/basic/test_scheduler_config.py
@@ -0,0 +1,96 @@
+import os
+import tempfile
+
+import yaml
+
+from cecli.args import get_parser
+from cecli.main import (
+    _scheduler_config,
+    _scheduler_enabled,
+    convert_yaml_to_json_string,
+)
+
+
+class _Args:
+    def __init__(self, scheduler=None):
+        self.scheduler = scheduler
+
+
+class _Workflow:
+    def __init__(self, schedule=None, enabled=True):
+        self.schedule = schedule
+        self.enabled = enabled
+
+
+def _parse_config(config, argv=()):
+    fd, path = tempfile.mkstemp(suffix=".yml")
+    os.close(fd)
+
+    try:
+        with open(path, "w") as f:
+            yaml.dump(config, f)
+
+        parser = get_parser([path], None)
+
+        return parser.parse_known_args(list(argv))[0]
+    finally:
+        os.unlink(path)
+
+
+def _resolve_scheduler(config, argv=()):
+    """Mirror main_async: parse the config, then deep-merge the scheduler arg."""
+
+    args = _parse_config(config, argv)
+
+    if args.scheduler is not None:
+        args.scheduler = convert_yaml_to_json_string(args.scheduler, config.get("scheduler"))
+
+    return _scheduler_config(args)
+
+
+def test_scheduler_object_parses_without_aborting_startup():
+    """Regression: the documented object form used to abort argparse startup."""
+
+    resolved = _resolve_scheduler({"scheduler": {"enabled": False, "tick_seconds": 5}})
+
+    assert resolved == {
+        "enabled": False,
+        "tick_seconds": 5,
+        "catch_up": True,
+        "max_catchup": 1,
+    }
+
+
+def test_scheduler_absent_uses_documented_defaults_and_auto():
+    resolved = _resolve_scheduler({})
+
+    assert resolved == {
+        "enabled": None,
+        "tick_seconds": 60,
+        "catch_up": True,
+        "max_catchup": 1,
+    }
+
+
+def test_cli_scheduler_json_overrides_config_per_key():
+    config = {"scheduler": {"enabled": True, "tick_seconds": 5}}
+    resolved = _resolve_scheduler(config, argv=["--scheduler", '{"enabled": false}'])
+
+    assert resolved["enabled"] is False  # CLI wins
+    assert resolved["tick_seconds"] == 5  # file-only key preserved
+
+
+def test_scheduler_config_defaults_on_bad_input():
+    assert _scheduler_config(_Args(None))["tick_seconds"] == 60
+    assert _scheduler_config(_Args("not json"))["tick_seconds"] == 60
+    assert _scheduler_config(_Args("[1, 2]"))["tick_seconds"] == 60
+
+
+def test_scheduler_enabled_explicit_and_auto():
+    scheduled = _Workflow(schedule="@daily")
+    manual = _Workflow(schedule=None)
+
+    assert _scheduler_enabled({"enabled": True}, [manual]) is True
+    assert _scheduler_enabled({"enabled": False}, [scheduled]) is False
+    assert _scheduler_enabled({"enabled": None}, [scheduled]) is True  # auto
+    assert _scheduler_enabled({"enabled": None}, [manual]) is False  # auto
diff --git a/tests/basic/test_watch.py b/tests/basic/test_watch.py
index 90d3bcb3195..388a3b3f33c 100644
--- a/tests/basic/test_watch.py
+++ b/tests/basic/test_watch.py
@@ -248,6 +248,34 @@ def test_catch_up_scan_ignores_file_at_threshold(tmp_path):
     assert watcher.changed_files == set()
 
 
+def test_catch_up_scan_advances_past_file_mtime(tmp_path):
+    """A file whose mtime is ahead of the wall clock is not re-found.
+
+    On Windows the wall clock (``time.time()``) can be coarser than the
+    filesystem timestamp, so a file written just before a scan may report an
+    mtime slightly ahead of the time the scan started. The threshold must be
+    advanced using the filesystem timestamp so the file is not re-found.
+    """
+    io = InputOutput(pretty=False, fancy_input=False, yes=False)
+    coder = MinimalCoder(io)
+    watcher = FileWatcher(coder, root=tmp_path)
+    watcher.last_scan_time = 0
+
+    target = tmp_path / "a.py"
+    target.write_text("# ai!\n")
+
+    # Simulate the clock skew by giving the file an mtime in the future
+    # relative to the wall clock reading taken when the scan starts.
+    ahead = time.time() + 5
+    os.utime(target, (ahead, ahead))
+
+    assert watcher.catch_up_scan()
+    watcher.changed_files = set()
+
+    assert not watcher.catch_up_scan()
+    assert watcher.changed_files == set()
+
+
 def test_process_changes_consumes_catch_up(tmp_path):
     io = InputOutput(pretty=False, fancy_input=False, yes=False)
     coder = MinimalCoder(io)
diff --git a/tests/helpers/observations/test_observation_service.py b/tests/helpers/observations/test_observation_service.py
index a8a438288b8..79a70aaece2 100644
--- a/tests/helpers/observations/test_observation_service.py
+++ b/tests/helpers/observations/test_observation_service.py
@@ -247,3 +247,48 @@ def fake_summarize(messages, prompt, max_tokens=None, coder=None):
     # The observation prompt is appended as the final user message
     assert sent_messages[-1]["role"] == "user"
     assert "Observation Prompt" in sent_messages[-1]["content"]
+
+
+@pytest.mark.asyncio
+async def test_compaction_bails_out_when_fixed_context_dominates():
+    """A large file in the chat cannot be compacted away, so bail instead of looping."""
+    from cecli.coders.base_coder import Coder
+
+    coder = MagicMock(spec=Coder)
+    coder.uuid = "test-coder-fixed-context"
+    coder.enable_context_compaction = True
+    coder.context_compaction_max_tokens = 1000
+    coder.context_compaction_summary_tokens = 100
+    coder.io = MagicMock()
+    coder.gpt_prompts = MagicMock()
+    coder._rate_limit_sleep = AsyncMock()
+
+    # done/cur/diff/file_context are empty, so the entire context is the fixed
+    # context (chat files, repo map, prompts). Summarizing cannot shrink it.
+    coder.summarizer = MagicMock()
+    coder.summarizer.count_tokens.side_effect = [0, 0, 0, 0, 6000]
+    coder.summarizer.summarize_all_as_text = AsyncMock(return_value="Summary Text")
+
+    mock_conv_manager = MagicMock()
+    mock_conv_manager.get_messages_dict.return_value = []
+
+    obs = MagicMock()
+    obs.check_and_trigger = AsyncMock()
+
+    with (
+        patch(
+            "cecli.coders.base_coder.ConversationService.get_manager",
+            return_value=mock_conv_manager,
+        ),
+        patch(
+            "cecli.coders.base_coder.ObservationService.get_instance",
+            return_value=obs,
+        ),
+    ):
+        await Coder.compact_context_if_needed(coder, force=True)
+
+    # Compaction must not summarize anything...
+    coder.summarizer.summarize_all_as_text.assert_not_called()
+    # ...and must tell the user why, with a remedy.
+    coder.io.tool_warning.assert_called_once()
+    coder.io.tool_output.assert_any_call("- Use /drop to remove unneeded files from the chat")
diff --git a/tests/helpers/scheduler/test_service.py b/tests/helpers/scheduler/test_service.py
index 4c20de93bde..1f14237d932 100644
--- a/tests/helpers/scheduler/test_service.py
+++ b/tests/helpers/scheduler/test_service.py
@@ -6,6 +6,7 @@
 """
 
 import asyncio
+import threading
 from datetime import datetime, timedelta, timezone
 from types import SimpleNamespace
 from unittest.mock import AsyncMock, MagicMock, patch
@@ -385,3 +386,63 @@ async def stubborn():
         task.cancel()
         with contextlib.suppress(asyncio.CancelledError):
             await task
+
+
+class TestCoderLoopBridge:
+    """Scheduled runs must execute on the coder's own (worker) event loop."""
+
+    def test_coder_loop_is_none_without_tui(self):
+        """A coder with no resolvable worker loop (linear/test) bridges to the current loop."""
+        service = _service([make_wf()])
+
+        assert service._coder_loop() is None
+
+    @pytest.mark.asyncio
+    async def test_on_coder_loop_awaits_directly_without_tui(self):
+        service = _service([make_wf()])
+        seen = []
+
+        async def make_coro():
+            seen.append(asyncio.get_running_loop())
+            return "ok"
+
+        result = await asyncio.wait_for(service._on_coder_loop(make_coro), timeout=5)
+
+        assert result == "ok"
+        assert seen == [asyncio.get_running_loop()]
+
+    @pytest.mark.asyncio
+    async def test_on_coder_loop_bridges_to_worker_loop(self):
+        """A scheduled dispatch must be awaited on the coder's worker-thread loop."""
+        service = _service([make_wf()])
+        foreign = asyncio.new_event_loop()
+        started = threading.Event()
+
+        def run_foreign():
+            asyncio.set_event_loop(foreign)
+            foreign.call_soon(started.set)
+            foreign.run_forever()
+
+        thread = threading.Thread(target=run_foreign, daemon=True)
+        thread.start()
+
+        try:
+            assert started.wait(timeout=5)
+            service.coder.tui = lambda: SimpleNamespace(worker=SimpleNamespace(loop=foreign))
+
+            assert service._coder_loop() is foreign
+
+            seen = []
+
+            async def make_coro():
+                seen.append(asyncio.get_running_loop())
+                return "bridged"
+
+            result = await asyncio.wait_for(service._on_coder_loop(make_coro), timeout=5)
+
+            assert result == "bridged"
+            assert seen == [foreign]
+        finally:
+            foreign.call_soon_threadsafe(foreign.stop)
+            thread.join(timeout=5)
+            foreign.close()
diff --git a/tests/helpers/test_llms_hyperscalers.py b/tests/helpers/test_llms_hyperscalers.py
index 036f16ead6d..e5264af9efc 100644
--- a/tests/helpers/test_llms_hyperscalers.py
+++ b/tests/helpers/test_llms_hyperscalers.py
@@ -153,6 +153,19 @@ def test_bedrock_endpoint_requires_region():
         raise AssertionError("expected ValueError without a region")
 
 
+def test_bedrock_endpoint_built_from_resolved_api_base(monkeypatch):
+    # The Converse URL is assembled from resolved["api_base"] verbatim rather
+    # than a hardcoded public host, so a custom endpoint (VPC endpoint, proxy,
+    # LocalStack, ...) resolved onto api_base is honored.
+    monkeypatch.setenv("AWS_REGION_NAME", "us-east-1")
+    resolved = {
+        "route": "anthropic.claude-sonnet",
+        "api_base": "https://bedrock.internal.example.com",
+    }
+    url = bedrock_endpoint(resolved)
+    assert url == ("https://bedrock.internal.example.com/model/anthropic.claude-sonnet/converse")
+
+
 # ---------------------------------------------------------------------------
 # Chat-family wiring: signer hook + extra_query
 # ---------------------------------------------------------------------------
diff --git a/tests/helpers/workflows/test_config.py b/tests/helpers/workflows/test_config.py
index 1aa38b1606f..0d017999634 100644
--- a/tests/helpers/workflows/test_config.py
+++ b/tests/helpers/workflows/test_config.py
@@ -44,6 +44,22 @@ def _mapping_text(name: str, prompt: str) -> str:
     return f"name: {name}\ntasks:\n  - prompt: {prompt}\n"
 
 
+def _local_dir_workflow(repo: Path, filename: str, text: str) -> Path:
+    wdir = repo / ".cecli" / "workflows"
+    wdir.mkdir(parents=True, exist_ok=True)
+    path = wdir / filename
+    path.write_text(text, encoding="utf-8")
+    return path
+
+
+def _agents_dir_workflow(repo: Path, filename: str, text: str) -> Path:
+    wdir = repo / ".agents" / "workflows"
+    wdir.mkdir(parents=True, exist_ok=True)
+    path = wdir / filename
+    path.write_text(text, encoding="utf-8")
+    return path
+
+
 # ── Case 1: single-mapping file, defaults ───────────────────────────────────
 
 
@@ -455,6 +471,85 @@ def test_bad_local_file_does_not_block_home_files(isolated, caplog):
 # ── Case 8: definition_hash stability ───────────────────────────────────────
 
 
+def test_local_dir_files_are_loaded(isolated):
+    _local_dir_workflow(isolated["repo"], "pipelines.yml", _mapping_text("local-dir", "check"))
+
+    configs = load_workflows(git_root=isolated["repo"])
+
+    assert [wf.name for wf in configs] == ["local-dir"]
+    assert configs[0].source_path == str(
+        isolated["repo"] / ".cecli" / "workflows" / "pipelines.yml"
+    )
+
+
+def test_local_dir_globs_stably_sorted(isolated):
+    _local_dir_workflow(isolated["repo"], "b_second.yml", _mapping_text("b", "second"))
+    _local_dir_workflow(isolated["repo"], "a_first.yml", _mapping_text("a", "first"))
+
+    configs = load_workflows(git_root=isolated["repo"])
+
+    assert [wf.name for wf in configs] == ["a", "b"]
+
+
+def test_local_dir_defaults_to_cwd(isolated):
+    _local_dir_workflow(isolated["repo"], "here.yml", _mapping_text("here", "cwd"))
+
+    configs = load_workflows()
+
+    assert [wf.name for wf in configs] == ["here"]
+
+
+def test_local_file_wins_over_local_dir(isolated):
+    _local_dir_workflow(isolated["repo"], "dir.yml", _mapping_text("dup", "dir"))
+    (isolated["repo"] / WORKFLOW_FILENAMES[0]).write_text(
+        _mapping_text("dup", "local-file"), encoding="utf-8"
+    )
+
+    configs = load_workflows(git_root=isolated["repo"])
+
+    assert [wf.tasks[0].prompt for wf in configs] == ["local-file"]
+
+
+def test_local_dir_wins_over_merged_config_and_home(isolated):
+    _local_dir_workflow(isolated["repo"], "dir.yml", _mapping_text("dup", "dir"))
+    _home_workflow(isolated["wdir"], "a_home.yml", _mapping_text("dup", "home"))
+    merged = {"workflows": [{"name": "dup", "tasks": [{"prompt": "merged"}]}]}
+
+    configs = load_workflows(merged_config=merged, git_root=isolated["repo"])
+
+    assert [wf.tasks[0].prompt for wf in configs] == ["dir"]
+
+
+def test_agents_dir_files_are_loaded(isolated):
+    _agents_dir_workflow(isolated["repo"], "pipelines.yml", _mapping_text("agents-dir", "check"))
+
+    configs = load_workflows(git_root=isolated["repo"])
+
+    assert [wf.name for wf in configs] == ["agents-dir"]
+    assert configs[0].source_path == str(
+        isolated["repo"] / ".agents" / "workflows" / "pipelines.yml"
+    )
+
+
+def test_cecli_dir_wins_over_agents_dir(isolated):
+    _agents_dir_workflow(isolated["repo"], "dup.yml", _mapping_text("dup", "agents"))
+    _local_dir_workflow(isolated["repo"], "dup.yml", _mapping_text("dup", "cecli"))
+
+    configs = load_workflows(git_root=isolated["repo"])
+
+    assert [wf.tasks[0].prompt for wf in configs] == ["cecli"]
+
+
+def test_agents_dir_wins_over_merged_config_and_home(isolated):
+    _agents_dir_workflow(isolated["repo"], "dir.yml", _mapping_text("dup", "agents"))
+    _home_workflow(isolated["wdir"], "a_home.yml", _mapping_text("dup", "home"))
+    merged = {"workflows": [{"name": "dup", "tasks": [{"prompt": "merged"}]}]}
+
+    configs = load_workflows(merged_config=merged, git_root=isolated["repo"])
+
+    assert [wf.tasks[0].prompt for wf in configs] == ["agents"]
+
+
 def test_definition_hash_stable_across_key_order():
     one = models.WorkflowConfig.from_dict(
         {"name": "h", "tasks": [{"prompt": "a", "name": "t1"}], "enabled": True}
diff --git a/tests/subagents/test_io_proxy.py b/tests/subagents/test_io_proxy.py
index 0b49b1d10e4..04501ab459e 100644
--- a/tests/subagents/test_io_proxy.py
+++ b/tests/subagents/test_io_proxy.py
@@ -185,3 +185,44 @@ async def test_setattr_forwards_to_target(self):
         proxy.some_attr = "value"
 
         assert target.some_attr == "value"
+
+    @pytest.mark.asyncio
+    async def test_confirm_ask_auto_approves_unattended_coder(self):
+        """A ``skip_cli_confirmations`` coder auto-approves without touching the target.
+
+        Regression: unattended ``wf:``/``ws:`` workers used to forward the
+        confirmation to the shared Textual IO and block on the input wake from a
+        foreign event loop, freezing the interactive coder.
+        """
+        from cecli.helpers.io_proxy import IOProxy
+
+        target = MagicMock()
+        target.confirm_ask = AsyncMock(return_value=False)
+
+        coder = MagicMock()
+        coder.uuid = "test-uuid"
+        coder.skip_cli_confirmations = True
+
+        proxy = IOProxy(target, coder)
+        result = await proxy.confirm_ask("Run tools?")
+
+        assert result is True
+        target.confirm_ask.assert_not_called()
+
+    @pytest.mark.asyncio
+    async def test_confirm_ask_forwards_for_interactive_coder(self):
+        """An interactive coder forwards ``confirm_ask`` with coder_uuid injected."""
+        from cecli.helpers.io_proxy import IOProxy
+
+        target = MagicMock()
+        target.confirm_ask = AsyncMock(return_value=True)
+
+        coder = MagicMock()
+        coder.uuid = "test-uuid"
+        coder.skip_cli_confirmations = False
+
+        proxy = IOProxy(target, coder)
+        result = await proxy.confirm_ask("Run tools?")
+
+        assert result is True
+        target.confirm_ask.assert_awaited_once_with("Run tools?", coder_uuid="test-uuid")
diff --git a/tests/subagents/test_service.py b/tests/subagents/test_service.py
index b1d836b79db..974af318a9c 100644
--- a/tests/subagents/test_service.py
+++ b/tests/subagents/test_service.py
@@ -2,6 +2,9 @@
 Tests for cecli/helpers/agents/service.py — AgentService.
 """
 
+import asyncio
+import threading
+import time
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
@@ -962,3 +965,108 @@ async def test_plain_worker_does_not_inherit(self, service):
         merged = await self._create(service, "worker")
 
         assert merged == {"own": 2}
+
+
+# ================================================================== #
+# Cross-loop lock safety
+# ================================================================== #
+
+
+class TestCrossLoopLocks:
+    """Spawn/conversation locks must be usable from two different event loops.
+
+    ``asyncio.Lock`` binds to the loop of its first *waiting* acquirer and
+    raises ``RuntimeError`` ("... is bound to a different event loop") when
+    awaited from another loop. Spawn, reap and conversation writes reach the
+    same per-parent lock from both the TUI main (Textual) loop and the coder
+    worker-thread loop, so the pools use ``threading.Lock``; the guarded
+    sections are synchronous, so a blocked acquirer simply waits.
+    """
+
+    def setup_method(self):
+        """Isolate the lock pools (class-level, so they leak between tests)."""
+        AgentService._spawn_locks = {}
+        AgentService._conversation_locks = {}
+
+    def teardown_method(self):
+        AgentService._spawn_locks = {}
+        AgentService._conversation_locks = {}
+
+    def test_lock_pools_are_thread_locks(self):
+        """The pools hand out ``threading.Lock``, never ``asyncio.Lock``."""
+        lock = AgentService._get_lock(AgentService._spawn_locks, "parent-uuid")
+        assert isinstance(lock, type(threading.Lock()))
+        assert not isinstance(lock, asyncio.Lock)
+        # Created lazily, but stable per uuid.
+        assert AgentService._get_lock(AgentService._spawn_locks, "parent-uuid") is lock
+
+    def test_contended_lock_is_usable_across_event_loops(self):
+        """Two waiters on their own loops must block, never raise.
+
+        With ``asyncio.Lock`` the *second* waiter (different loop from the one
+        the lock bound to) raises ``RuntimeError``; with ``threading.Lock``
+        both simply wait for the holder to release.
+        """
+        lock = AgentService._get_lock(AgentService._spawn_locks, "parent-uuid")
+
+        entered = threading.Event()
+        release = threading.Event()
+        order = []
+        errors = []
+
+        def run_on_own_loop(coro_factory):
+            """Run the coroutine on a fresh event loop in this thread."""
+            loop = asyncio.new_event_loop()
+            try:
+                asyncio.set_event_loop(loop)
+                loop.run_until_complete(coro_factory())
+            except BaseException as exc:  # noqa: BLE001 - record, do not raise in thread
+                errors.append(exc)
+            finally:
+                asyncio.set_event_loop(None)
+                loop.close()
+
+        async def holds_lock():
+            with lock:
+                order.append("holder")
+                entered.set()
+                # Await *outside* the lock: guarded sections never await.
+                await asyncio.to_thread(release.wait, 5)
+            order.append("holder-released")
+
+        def contender_factory(tag):
+            async def contends():
+                with lock:
+                    order.append(tag)
+
+            return contends
+
+        holder = threading.Thread(target=run_on_own_loop, args=(holds_lock,))
+        holder.start()
+        first = second = None
+        try:
+            assert entered.wait(timeout=5), "holder never acquired the lock"
+
+            # Waiter #1 blocks on its own loop...
+            first = threading.Thread(target=run_on_own_loop, args=(contender_factory("c1"),))
+            first.start()
+            time.sleep(0.1)
+            # ...waiter #2 on a third loop must block too, not raise.
+            second = threading.Thread(target=run_on_own_loop, args=(contender_factory("c2"),))
+            second.start()
+            time.sleep(0.2)
+
+            assert errors == [], f"cross-loop lock acquisition raised: {errors}"
+            assert order == ["holder"]
+        finally:
+            release.set()
+            holder.join(timeout=5)
+
+        for thread in (first, second):
+            if thread is not None:
+                thread.join(timeout=5)
+
+        assert errors == []
+        assert "c1" in order
+        assert "c2" in order
+        assert "holder-released" in order
diff --git a/tests/unit/test_queues.py b/tests/unit/test_queues.py
index 3e75460a254..af2c38f506f 100644
--- a/tests/unit/test_queues.py
+++ b/tests/unit/test_queues.py
@@ -122,12 +122,14 @@ def setup_method(self):
         queues._per_coder_queues.clear()
         queues._input_loop = None
         queues._input_wake = None
+        queues._input_wakes.clear()
 
     def teardown_method(self):
         """Reset the global input-loop state after each test."""
         queues._per_coder_queues.clear()
         queues._input_loop = None
         queues._input_wake = None
+        queues._input_wakes.clear()
 
     def test_wake_with_closed_loop_drops_stale_binding(self):
         """Waking after the bound loop closed must not raise, and resets state."""
@@ -176,3 +178,65 @@ async def run():
                 await task
 
         asyncio.run(run())
+
+    def test_foreign_live_loop_does_not_steal_primary_binding(self):
+        """A consumer on a foreign *live* loop must not steal the primary binding.
+
+        Regression for the scheduler freeze: a scheduled workflow running on the
+        main (Textual) loop used to call wait_for_input() and rebind the global
+        wake to its own loop, stranding the primary coder's input pump that was
+        parked on the worker-thread loop — with no way to recover.
+        """
+        import threading
+
+        ready = threading.Event()
+        released = threading.Event()
+        loop_a = {}
+
+        async def park():
+            ready.set()
+            await queues.wait_for_input()
+            released.set()
+
+        def run_loop_a():
+            loop = asyncio.new_event_loop()
+            asyncio.set_event_loop(loop)
+            loop_a["loop"] = loop
+            queues.set_input_loop(loop)
+            loop.create_task(park())
+            loop.run_forever()
+
+        thread = threading.Thread(target=run_loop_a, daemon=True)
+        thread.start()
+
+        try:
+            assert ready.wait(timeout=5)
+            assert loop_a["loop"] is queues._input_loop
+
+            async def main():
+                primary = queues._input_loop
+                assert primary is loop_a["loop"]
+                assert not primary.is_closed()
+
+                task = asyncio.create_task(queues.wait_for_input())
+                await asyncio.sleep(0)
+
+                # The foreign loop must NOT rebind the primary wake slot...
+                assert queues._input_loop is primary
+                # ...it gets a dedicated wake event instead.
+                assert asyncio.get_running_loop() in queues._input_wakes
+
+                queues.wake_input_waiters()
+                await asyncio.wait_for(task, timeout=5)
+
+            asyncio.run(main())
+
+            # Both the primary (loop A) and the foreign waiter were released.
+            assert released.wait(timeout=5)
+        finally:
+            loop = loop_a.get("loop")
+            if loop is not None and not loop.is_closed():
+                loop.call_soon_threadsafe(loop.stop)
+            thread.join(timeout=5)
+            if loop is not None and not loop.is_closed():
+                loop.close()