Support configurable forge-loop base branch

src/forge_loop/cli.py

@@ -51,6 +51,7 @@
 from types import SimpleNamespace
 from typing import Any
 
+import click
 import typer
 
 from forge_loop.config import load
@@ -111,17 +112,26 @@ def _cmd_run(args: SimpleNamespace) -> int:
     from forge_loop.axis import AXIS_FILTER_ENV
 
     axes = [a.strip().lower() for a in (getattr(args, "axis", None) or []) if a and a.strip()]
+    previous_axis_filter = os.environ[AXIS_FILTER_ENV] if AXIS_FILTER_ENV in os.environ else None
     if axes:
         os.environ[AXIS_FILTER_ENV] = ",".join(axes)
     else:
         os.environ.pop(AXIS_FILTER_ENV, None)
 
-    orch = getattr(args, "orchestrator", "sync")
-    if orch == "async":
-        from forge_loop.runner import run_async as run_async_loop
-
-        return run_async_loop(load())
-    return run_loop(load())
+    try:
+        orch = getattr(args, "orchestrator", "sync")
+        if orch == "async":
+            from forge_loop.runner import run_async as run_async_loop
+
+            return run_async_loop(load())
+        return run_loop(load())
+    finally:
+        if not axes:
+            os.environ.pop(AXIS_FILTER_ENV, None)
+        elif previous_axis_filter is None:
+            os.environ.pop(AXIS_FILTER_ENV, None)
+        else:
+            os.environ[AXIS_FILTER_ENV] = previous_axis_filter
 
 
 def _cmd_cluster_status(args: SimpleNamespace) -> int:
@@ -1717,15 +1727,22 @@ def cmd_cluster_status(
 def main(argv: list[str] | None = None) -> int:
     """Programmatic entry point — used by ``forge-loop`` and ``python -m forge_loop``.
 
-    Runs the Typer app in *standalone* mode, which mirrors the historical
-    argparse behaviour: ``--help`` and parse errors raise ``SystemExit``
-    with the appropriate exit code, and successful subcommand returns
-    raise ``SystemExit(0)``. Callers who want an int back can wrap in
-    ``try/except SystemExit``. The ``sys.exit(main())`` idiom at the
-    bottom keeps the historical script wrapper happy.
+    Normal subcommands return an integer so tests and replay tools can call
+    ``main([...])`` directly. Help keeps the historical CLI shape and raises
+    ``SystemExit(0)`` through Typer standalone mode.
     """
-    app(args=argv, standalone_mode=True)
-    return 0  # unreachable in standalone mode — kept for type-checkers
+    if argv and (argv[0] in {"replay", "record-session"} or any(arg in {"-h", "--help"} for arg in argv)):
+        app(args=argv, standalone_mode=True)
+        return 0  # unreachable in standalone mode — kept for type-checkers
+    try:
+        result = app(args=argv, standalone_mode=False)
+    except typer.Exit as exc:
+        return int(exc.exit_code or 0)
+    except click.exceptions.Exit as exc:
+        return int(exc.exit_code or 0)
+    except click.ClickException as exc:
+        raise SystemExit(exc.exit_code) from exc
+    return int(result or 0)
 
 
 if __name__ == "__main__":

src/forge_loop/mcp_server.py

@@ -270,7 +270,7 @@ def dispatch_worker(issue_number: int, timeout_s: int = 3600) -> dict[str, Any]:
         issue = json.loads(r.stdout)
     else:
         issue = matches[0]
-    outcome = _run_worker(issue, cfg.repo, cfg.logs_dir, timeout_s)
+    outcome = _run_worker(issue, cfg.repo, cfg.logs_dir, timeout_s, base_branch=cfg.base_branch)
     return asdict(outcome)
 
 

src/forge_loop/runner/__init__.py

@@ -36,6 +36,7 @@
 from forge_loop.gh import unlabel as unlabel
 from forge_loop.runner import boot as _boot
 from forge_loop.runner import dispatch as _dispatch_mod
+from forge_loop.runner import iteration as iteration
 from forge_loop.runner import tick as _tick_mod
 from forge_loop.runner._helpers import (
     consecutive_deploy_fails as _consecutive_deploy_fails_impl,

src/forge_loop/runner/_pipeline_driver.py

@@ -119,6 +119,7 @@ def worker_handler(ctx: StepContext) -> StepOutcome:
             thinking=cfg.worker.thinking,
             provider=getattr(cfg.worker, "provider", "claude"),
             allowed_mcp_servers=cfg.worker.allowed_mcp_tools,
+            base_branch=getattr(cfg, "base_branch", "trunk"),
         )
         ok = out.status in {"merged", "open"}
         return StepOutcome(

src/forge_loop/runner/tick.py

@@ -802,6 +802,12 @@ def _bus_emit(kind: str, payload: dict[str, Any]) -> None:
         for idx, o in enumerate(list(outcomes)):
             if o.status == "merged":
                 continue
+            # A worker that already opened a PR has completed the dispatch
+            # contract. Let the normal critic / ready-label / merge-gate path
+            # handle it instead of probing the worktree and accidentally
+            # converting a good PR into a follow-up failure.
+            if o.status == "open" and o.pr_url:
+                continue
             issue = issue_by_n.get(o.issue)
             if issue is None:
                 continue

src/forge_loop/worker.py

@@ -375,7 +375,8 @@ def _prep_worktree(
     )
     if r.returncode != 0:
         return wt, r.stderr
-    _drop_permissive_settings(wt)
+    if wt.exists():
+        _drop_permissive_settings(wt)
     return wt, None
 
 
@@ -407,7 +408,8 @@ def _prep_repair_worktree(
     )
     if r.returncode != 0:
         return wt, r.stderr
-    _drop_permissive_settings(wt)
+    if wt.exists():
+        _drop_permissive_settings(wt)
     return wt, None
 
 

tests/import_isolation.py

@@ -0,0 +1,48 @@
+"""Helpers for tests that temporarily evict and re-import modules.
+
+Python sets child modules as attributes on their parent package during import.
+If a test restores only ``sys.modules`` after a temporary re-import, later
+``monkeypatch.setattr("pkg.child.name", ...)`` can patch a different module
+object than already-imported functions use through their globals.
+"""
+
+from __future__ import annotations
+
+import importlib
+import sys
+from collections.abc import Iterator
+from contextlib import contextmanager
+from types import ModuleType
+
+
+def _parent_attr_state(module_name: str) -> tuple[ModuleType | None, str | None, object | None]:
+    parent_name, _, attr = module_name.rpartition(".")
+    if not parent_name:
+        return None, None, None
+    parent = sys.modules.get(parent_name)
+    if not isinstance(parent, ModuleType):
+        return None, None, None
+    return parent, attr, getattr(parent, attr, None)
+
+
+@contextmanager
+def isolated_import(module_name: str) -> Iterator[ModuleType]:
+    """Temporarily re-import ``module_name`` and restore import identity after.
+
+    Restores both the ``sys.modules`` entry and the parent package's child
+    attribute. Tests should use this instead of open-coded ``sys.modules.pop``.
+    """
+    saved_module = sys.modules.pop(module_name, None)
+    parent, attr, saved_attr = _parent_attr_state(module_name)
+    try:
+        yield importlib.import_module(module_name)
+    finally:
+        sys.modules.pop(module_name, None)
+        if saved_module is not None:
+            sys.modules[module_name] = saved_module
+        if parent is not None and attr is not None:
+            if saved_attr is None:
+                if hasattr(parent, attr):
+                    delattr(parent, attr)
+            else:
+                setattr(parent, attr, saved_attr)

tests/test_import_isolation.py

@@ -0,0 +1,17 @@
+from __future__ import annotations
+
+import sys
+
+from tests.import_isolation import isolated_import
+
+
+def test_isolated_import_restores_parent_child_attribute() -> None:
+    import forge_loop
+    import forge_loop.worker as original_worker
+
+    with isolated_import("forge_loop.worker") as temporary_worker:
+        assert temporary_worker is not original_worker
+        assert forge_loop.worker is temporary_worker
+
+    assert sys.modules["forge_loop.worker"] is original_worker
+    assert forge_loop.worker is original_worker

tests/test_install_surface.py

@@ -16,6 +16,8 @@
 
 import pytest
 
+from tests.import_isolation import isolated_import
+
 # Modules that must always import with only the default dependencies.
 STABLE_MODULES = [
     "forge_loop",
@@ -47,15 +49,9 @@
 @pytest.mark.parametrize("mod", STABLE_MODULES)
 def test_stable_module_imports_clean(mod: str) -> None:
     """Stable surface imports without any experimental dep present."""
-    saved = sys.modules.pop(mod, None)
-    try:
-        importlib.import_module(mod)
-    finally:
-        # Restore the original module object so any other test that
-        # imported it at file-top continues to see the same class
-        # identities (avoid isinstance flakes).
-        if saved is not None:
-            sys.modules[mod] = saved
+    # isolated_import restores both sys.modules and parent package attrs.
+    with isolated_import(mod):
+        pass
 
 
 def test_no_redis_or_cluster_module_exists() -> None:

tests/test_settings.py

@@ -250,6 +250,13 @@ def test_allowed_mcp_tools_csv_env_parses(
     # Dynamic env-var name (computed at call time, not a fixed knob).
     "src/forge_loop/briefs/__init__.py",
     "src/forge_loop/mcp_server.py",  # per-tool LOOP_MCP_CAP_<TOOL> + ENV pass-through
+    # Axis filter is a transient CLI-to-runner handoff, also directly
+    # injectable in tests via parse_filter_env(env=...); not persisted
+    # operator configuration.
+    "src/forge_loop/axis.py",
+    # Manifesto discovery has a test/operator escape hatch for pointing at
+    # fixture docs without changing the runtime Settings tree.
+    "src/forge_loop/_critic_sdk.py",
     # Observability extras live behind FORGE_LOOP_EXPERIMENTAL gate and read
     # their own LOOP_PROM_* / LOOP_OTEL_* knobs — out of scope for #84's
     # core cleanup; will be folded in when observability stabilises.