diff --git a/tests/test_codex_loop.py b/tests/test_codex_loop.py
new file mode 100644
index 0000000..410fe7d
--- /dev/null
+++ b/tests/test_codex_loop.py
@@ -0,0 +1,289 @@
+import importlib
+import inspect
+import signal
+import sys
+import unittest
+from io import StringIO
+from unittest.mock import patch
+
+from the_lab import agent_cli
+from the_lab.agent_cli import _build_launch_command
+
+
+def _loop_module():
+    try:
+        return importlib.import_module("the_lab.codex_loop")
+    except ModuleNotFoundError as error:
+        raise AssertionError("Codex loop runner is missing") from error
+
+
+class CodexLoopTests(unittest.TestCase):
+    def test_claude_keeps_native_loop_prompt(self):
+        command = _build_launch_command(
+            "claude",
+            "claude",
+            "/loop 5m keep working",
+            None,
+            False,
+        )
+
+        self.assertEqual(
+            command,
+            ["claude", "--dangerously-skip-permissions", "/loop 5m keep working"],
+        )
+
+    def test_codex_loop_launches_scheduler_without_slash_command(self):
+        parameters = inspect.signature(_build_launch_command).parameters
+        if "loop_duration" not in parameters:
+            self.fail("Codex launch command has no loop scheduler support")
+
+        command = _build_launch_command(
+            "codex",
+            "codex",
+            "keep working",
+            "gpt-test",
+            False,
+            extra_agent_args=["--search"],
+            loop_duration="5m",
+        )
+
+        self.assertEqual(command[:3], [sys.executable, "-m", "the_lab.codex_loop"])
+        self.assertIn("5m", command)
+        self.assertIn("keep working", command)
+        self.assertNotIn("/loop", " ".join(command))
+
+    def test_only_codex_loop_disables_interactive_child_input(self):
+        if not hasattr(agent_cli, "_uses_interactive_child"):
+            self.fail("agent CLI cannot distinguish the headless Codex loop")
+
+        self.assertFalse(agent_cli._uses_interactive_child("codex", True))
+        self.assertTrue(agent_cli._uses_interactive_child("codex", False))
+        self.assertTrue(agent_cli._uses_interactive_child("claude", True))
+
+    def test_advertised_durations_map_to_seconds(self):
+        normalise = _loop_module()._normalise_duration
+
+        self.assertEqual(normalise("30s"), 30)
+        self.assertEqual(normalise("61s"), 61)
+        self.assertEqual(normalise("5m"), 300)
+        self.assertEqual(normalise("2h"), 7_200)
+        self.assertEqual(normalise("1d"), 86_400)
+
+    def test_turn_runs_immediately_then_waits_remaining_interval(self):
+        module = _loop_module()
+        times = iter((100.0, 220.0, 400.0))
+        events = []
+        turns = []
+
+        class Stop:
+            stopped = False
+
+            def is_set(self):
+                return self.stopped
+
+            def wait(self, delay):
+                events.append(("wait", delay))
+                return False
+
+        stop = Stop()
+
+        def run_turn(thread_id):
+            turns.append(thread_id)
+            events.append(("run", thread_id))
+            if len(turns) == 2:
+                stop.stopped = True
+            return "thread-123"
+
+        module._run_schedule(300, run_turn, stop, lambda: next(times))
+
+        self.assertEqual(
+            events,
+            [("run", None), ("wait", 180), ("run", "thread-123")],
+        )
+
+    def test_turn_commands_resume_the_exact_thread_and_repeat_the_prompt(self):
+        build = _loop_module()._build_turn_command
+        prompt = "run the next experiment"
+
+        initial = build("codex", "gpt-test", ["--search"], None, prompt)
+        resumed = build("codex", "gpt-test", ["--search"], "thread-123", prompt)
+
+        self.assertEqual(
+            initial,
+            [
+                "codex",
+                "--search",
+                "exec",
+                "--dangerously-bypass-approvals-and-sandbox",
+                "--model",
+                "gpt-test",
+                prompt,
+            ],
+        )
+        self.assertEqual(
+            resumed,
+            [
+                "codex",
+                "--search",
+                "exec",
+                "resume",
+                "--dangerously-bypass-approvals-and-sandbox",
+                "--model",
+                "gpt-test",
+                "thread-123",
+                prompt,
+            ],
+        )
+
+    def test_long_turn_runs_once_immediately_without_catch_up_queue(self):
+        module = _loop_module()
+        times = iter((100.0, 800.0, 800.0, 810.0))
+        waits = []
+        turns = []
+
+        class Stop:
+            stopped = False
+
+            def is_set(self):
+                return self.stopped
+
+            def wait(self, delay):
+                waits.append(delay)
+                return len(waits) == 2
+
+        stop = Stop()
+
+        def run_turn(thread_id):
+            turns.append(thread_id)
+            return "thread-123"
+
+        module._run_schedule(300, run_turn, stop, lambda: next(times))
+
+        self.assertEqual(waits, [0, 290])
+        self.assertEqual(turns, [None, "thread-123"])
+
+    def test_stop_during_wait_prevents_another_turn(self):
+        module = _loop_module()
+        turns = []
+        waits = []
+
+        class Stop:
+            stopped = False
+
+            def is_set(self):
+                return self.stopped
+
+            def wait(self, delay):
+                waits.append(delay)
+                self.stopped = True
+                return False
+
+        stop = Stop()
+
+        module._run_schedule(
+            300,
+            lambda thread_id: turns.append(thread_id) or "thread-123",
+            stop,
+            iter((100.0, 101.0)).__next__,
+        )
+
+        self.assertEqual(turns, [None])
+        self.assertEqual(waits, [299])
+
+    def test_pending_signal_is_replayed_to_new_child(self):
+        module = _loop_module()
+        if not hasattr(module, "_activate_process"):
+            self.fail("Codex loop cannot replay a signal received during spawn")
+
+        class Process:
+            def __init__(self):
+                self.terminated = 0
+
+            def poll(self):
+                return None
+
+            def terminate(self):
+                self.terminated += 1
+
+        process = Process()
+        module._stop.set()
+        module._signal_received = signal.SIGINT
+        self.addCleanup(module._stop.clear)
+        self.addCleanup(setattr, module, "_signal_received", None)
+        self.addCleanup(setattr, module, "_active_process", None)
+
+        module._activate_process(process)
+
+        self.assertEqual(process.terminated, 1)
+
+    def test_ctrl_c_terminates_active_child(self):
+        module = _loop_module()
+
+        class Process:
+            terminated = 0
+
+            def poll(self):
+                return None
+
+            def terminate(self):
+                self.terminated += 1
+
+        process = Process()
+        module._active_process = process
+        module._stop.clear()
+        self.addCleanup(module._stop.clear)
+        self.addCleanup(setattr, module, "_signal_received", None)
+        self.addCleanup(setattr, module, "_active_process", None)
+
+        module._handle_signal(signal.SIGINT, None)
+
+        self.assertTrue(module._stop.is_set())
+        self.assertEqual(process.terminated, 1)
+
+    def test_run_turn_forwards_native_output_and_reads_session_id(self):
+        module = _loop_module()
+        lines = [
+            "OpenAI Codex v0.149.0\n",
+            "session id: 01a02689-b39b-7cd1-ac1d-5e32de2ab114\n",
+            "codex\n",
+        ]
+
+        class Process:
+            stderr = iter(lines)
+
+            def poll(self):
+                return None
+
+            def wait(self):
+                return 0
+
+        output = StringIO()
+        with (
+            patch.object(module.subprocess, "Popen", return_value=Process()) as popen,
+            patch.object(sys, "stderr", output),
+        ):
+            thread_id = module._run_turn(["codex"], None)
+
+        self.assertEqual(thread_id, "01a02689-b39b-7cd1-ac1d-5e32de2ab114")
+        self.assertEqual(output.getvalue(), "".join(lines))
+        self.assertIs(popen.call_args.kwargs["stderr"], module.subprocess.PIPE)
+        self.assertTrue(popen.call_args.kwargs["start_new_session"])
+        self.assertNotIn("stdin", popen.call_args.kwargs)
+        self.assertNotIn("stdout", popen.call_args.kwargs)
+
+    def test_thread_id_is_read_from_native_codex_header(self):
+        extract = _loop_module()._thread_id_from_line
+
+        self.assertEqual(
+            extract("session id: 01a02689-b39b-7cd1-ac1d-5e32de2ab114"),
+            "01a02689-b39b-7cd1-ac1d-5e32de2ab114",
+        )
+        self.assertIsNone(extract("session id:"))
+        self.assertIsNone(extract("session id: ----"))
+        self.assertIsNone(extract("session id: deadbeef"))
+        self.assertIsNone(extract("session id: a-b-c"))
+        self.assertIsNone(extract("not a session header"))
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/the_lab/agent_cli.py b/the_lab/agent_cli.py
index eda3452..6ada159 100644
--- a/the_lab/agent_cli.py
+++ b/the_lab/agent_cli.py
@@ -68,6 +68,10 @@ def _agent_binary(agent: str) -> str:
     return shutil.which(name) or name
 
 
+def _uses_interactive_child(agent: str, use_loop: bool) -> bool:
+    return not (agent == "codex" and use_loop)
+
+
 def _build_launch_command(
     agent: str,
     agent_bin: str,
@@ -78,6 +82,7 @@ def _build_launch_command(
     mcp_path: str | None = None,
     sandboxed: bool = False,
     extra_agent_args: list[str] | None = None,
+    loop_duration: str | None = None,
 ) -> list[str]:
     """Build the agent launch command.
 
@@ -92,6 +97,7 @@ def _build_launch_command(
     extra_agent_args: additional flags forwarded verbatim to the agent binary,
                 inserted before the ``--`` / prompt separator so they are
                 parsed by the agent itself.  Example: ``['--resume', '<id>']``.
+    loop_duration: run Codex through its fixed-schedule resume loop when set.
     """
     if agent == "claude":
         cmd = [agent_bin]
@@ -111,6 +117,24 @@ def _build_launch_command(
         cmd.append(loop_prompt)
         return cmd
 
+    if loop_duration is not None:
+        cmd = [
+            sys.executable,
+            "-m",
+            "the_lab.codex_loop",
+            "--codex-bin",
+            agent_bin,
+            "--duration",
+            loop_duration,
+            "--prompt",
+            loop_prompt,
+        ]
+        if model:
+            cmd.extend(["--model", model])
+        if extra_agent_args:
+            cmd.extend(["--", *extra_agent_args])
+        return cmd
+
     cmd = [agent_bin, "--yolo"]
     if model:
         cmd.extend(["--model", model])
@@ -346,7 +370,7 @@ def main():
         }}})
         print(f"MCP bridge: labapi → {api_base}", file=sys.stderr)
 
-    # Build the prompt argument for Claude
+    # Build the prompt argument for the selected agent.
     role_suffix = f" (role='{effective_role}')" if effective_role != "default" else ""
     role_arg = f"role='{effective_role}'" if effective_role != "default" else ""
     if use_loop:
@@ -356,14 +380,18 @@ def main():
         #     left off. Only re-read instructions if genuinely lost (e.g. after
         #     a crash/restart or when unsure what to do next).
         tool_call = f"get_instructions(role='{effective_role}')" if role_arg else "get_instructions()"
-        agent_prompt = (
-            f"/loop {args.duration} "
+        loop_task = (
             f"Start by calling {tool_call} to load the project instructions and API reference. "
             f"Then enter a continuous optimisation loop: propose and run experiments, "
             f"analyse results, and keep improving. "
             f"Only call {tool_call} again if you have lost context and are unsure what to do — "
             f"otherwise trust your current context and keep working."
         )
+        agent_prompt = (
+            f"/loop {args.duration} {loop_task}"
+            if args.agent == "claude"
+            else loop_task
+        )
         print(f"Mode: loop (every {args.duration}){role_suffix}", file=sys.stderr)
     else:
         # Prepend a directive to call get_instructions first if MCP is available
@@ -446,6 +474,7 @@ def main():
         mcp_path=mcp_path_in_sandbox,  # None when not sandboxed → writes to /tmp itself
         sandboxed=bool(sandbox_mode),
         extra_agent_args=extra_agent_args or None,
+        loop_duration=args.duration if use_loop and args.agent == "codex" else None,
     )
 
     env = dict(os.environ)
@@ -967,9 +996,10 @@ def _sample_screen():
         # We set stdin to raw mode so every keypress (arrows, ctrl-*, etc.)
         # is forwarded immediately rather than line-buffered.
         import tty as _tty
+        _interactive_child = _uses_interactive_child(args.agent, use_loop)
         _old_tty: list | None = None
         try:
-            if sys.stdin.isatty():
+            if _interactive_child and sys.stdin.isatty():
                 _old_tty = _termios.tcgetattr(sys.stdin.fileno())
                 _tty.setraw(sys.stdin.fileno(), _termios.TCSANOW)
         except Exception:
@@ -1007,8 +1037,9 @@ def _stdin_relay():
             finally:
                 pass  # TTY restored in the main finally block
 
-        _t_stdin = _threading.Thread(target=_stdin_relay, daemon=True)
-        _t_stdin.start()
+        if _interactive_child:
+            _t_stdin = _threading.Thread(target=_stdin_relay, daemon=True)
+            _t_stdin.start()
 
         forwarded = {"sig": None}
 
diff --git a/the_lab/codex_loop.py b/the_lab/codex_loop.py
new file mode 100644
index 0000000..d5deede
--- /dev/null
+++ b/the_lab/codex_loop.py
@@ -0,0 +1,174 @@
+"""Run recurring Codex turns with a minimum start-to-start interval."""
+
+import argparse
+import re
+import signal
+import subprocess
+import sys
+import threading
+import time
+from collections.abc import Callable
+
+_DURATION_RE = re.compile(r"^(\d+)([smhd])$")
+_SESSION_ID_RE = re.compile(
+    r"^session id:\s*"
+    r"([0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-"
+    r"[0-9a-fA-F]{4}-[0-9a-fA-F]{12})\s*$"
+)
+_UNIT_SECONDS = {"s": 1, "m": 60, "h": 3600, "d": 86400}
+
+
+def _normalise_duration(value: str) -> int:
+    match = _DURATION_RE.fullmatch(value)
+    if match is None or int(match.group(1)) < 1:
+        raise ValueError("duration must be a positive integer followed by s, m, h, or d")
+    return int(match.group(1)) * _UNIT_SECONDS[match.group(2)]
+
+
+def _build_turn_command(
+    codex_bin: str,
+    model: str | None,
+    extra_args: list[str],
+    thread_id: str | None,
+    prompt: str,
+) -> list[str]:
+    command = [codex_bin, *extra_args, "exec"]
+    if thread_id is not None:
+        command.append("resume")
+    command.append("--dangerously-bypass-approvals-and-sandbox")
+    if model:
+        command.extend(["--model", model])
+    if thread_id is not None:
+        command.append(thread_id)
+    command.append(prompt)
+    return command
+
+
+def _thread_id_from_line(line: str) -> str | None:
+    match = _SESSION_ID_RE.fullmatch(line.strip())
+    return match.group(1) if match else None
+
+
+def _run_schedule(
+    interval_seconds: int,
+    run_turn: Callable[[str | None], str],
+    stop: threading.Event,
+    clock: Callable[[], float] = time.monotonic,
+) -> None:
+    thread_id = None
+    while not stop.is_set():
+        started = clock()
+        thread_id = run_turn(thread_id)
+        if stop.is_set():
+            break
+        elapsed = clock() - started
+        if stop.wait(max(0, interval_seconds - elapsed)):
+            break
+
+
+_active_process: subprocess.Popen[str] | None = None
+_stop = threading.Event()
+_signal_received: int | None = None
+
+
+def _terminate_process(process: subprocess.Popen[str]) -> None:
+    try:
+        if process.poll() is None:
+            process.terminate()
+    except OSError:
+        pass
+
+
+def _activate_process(process: subprocess.Popen[str]) -> None:
+    global _active_process
+    _active_process = process
+    if _stop.is_set():
+        _terminate_process(process)
+
+
+def _run_turn(command: list[str], expected_thread_id: str | None) -> str:
+    if _stop.is_set():
+        raise InterruptedError("Codex loop was stopped before the turn started")
+    process = subprocess.Popen(
+        command,
+        stderr=subprocess.PIPE,
+        text=True,
+        bufsize=1,
+        start_new_session=True,
+    )
+    _activate_process(process)
+    thread_id = None
+    try:
+        if process.stderr is None:
+            raise RuntimeError("Codex loop could not open its error pipe")
+        for line in process.stderr:
+            sys.stderr.write(line)
+            sys.stderr.flush()
+            thread_id = _thread_id_from_line(line) or thread_id
+        return_code = process.wait()
+    finally:
+        _active_process = None
+    if return_code:
+        raise subprocess.CalledProcessError(return_code, command)
+    if thread_id is None:
+        raise RuntimeError("Codex did not report a thread id")
+    if expected_thread_id is not None and thread_id != expected_thread_id:
+        raise RuntimeError("Codex resumed a different thread")
+    return thread_id
+
+
+def _handle_signal(signum: int, _frame) -> None:
+    global _signal_received
+    _signal_received = signum
+    _stop.set()
+    if _active_process is not None:
+        _terminate_process(_active_process)
+
+
+def main(argv: list[str] | None = None) -> int:
+    global _signal_received
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--codex-bin", required=True)
+    parser.add_argument("--duration", required=True)
+    parser.add_argument("--prompt", required=True)
+    parser.add_argument("--model")
+    parser.add_argument("codex_args", nargs=argparse.REMAINDER)
+    args = parser.parse_args(argv)
+    extra_args = args.codex_args[1:] if args.codex_args[:1] == ["--"] else args.codex_args
+    try:
+        interval_seconds = _normalise_duration(args.duration)
+    except ValueError as error:
+        parser.error(str(error))
+
+    _stop.clear()
+    _signal_received = None
+    for signum in (signal.SIGINT, signal.SIGTERM, signal.SIGHUP):
+        signal.signal(signum, _handle_signal)
+
+    print(f"Codex loop minimum interval: {args.duration}", file=sys.stderr)
+
+    def run_turn(thread_id: str | None) -> str:
+        command = _build_turn_command(
+            args.codex_bin,
+            args.model,
+            extra_args,
+            thread_id,
+            args.prompt,
+        )
+        return _run_turn(command, thread_id)
+
+    try:
+        _run_schedule(interval_seconds, run_turn, _stop)
+    except subprocess.CalledProcessError as error:
+        if not _stop.is_set():
+            print(f"Codex turn failed with status {error.returncode}", file=sys.stderr)
+            return error.returncode
+    except (OSError, RuntimeError) as error:
+        if not _stop.is_set():
+            print(f"Codex loop failed: {error}", file=sys.stderr)
+            return 1
+    return 128 + _signal_received if _signal_received is not None else 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())