Add forced one-shot scheduler runs

Allow codex-bg once --force to bypass workspace and plugin due intervals for smoke tests without bypassing dedupe, issue age filters, rate limits, or dry-run safety. Also allow the Codex pre-screener to run from its temporary artifact directory by passing --skip-git-repo-check.

Validation: .venv/bin/pre-commit run --all-files; .venv/bin/ruff check .; .venv/bin/bandit -c pyproject.toml -r src; git ls-files -z | xargs -0 .venv/bin/detect-secrets-hook --baseline .secrets.baseline; .venv/bin/pip-audit -r requirements-dev.txt; .venv/bin/actionlint; .venv/bin/zizmor --offline .github; PYTHONPATH=src python3 -m unittest discover -s tests; python3 -m compileall -q src tests; .venv/bin/python -m build; deployed to /home/rger/proj/codex-bg-test and ran forced dry-run.

Author: rgerhards

src/codex_bg/cli.py

@@ -20,6 +20,8 @@ def main(argv: list[str] | None = None) -> int:
         subparser.add_argument("--config", default=argparse.SUPPRESS)
         subparser.add_argument("--debug", action="store_true", default=argparse.SUPPRESS)
         subparser.add_argument("--dry-run", action="store_true", default=argparse.SUPPRESS)
+        if command == "once":
+            subparser.add_argument("--force", action="store_true")
     args = parser.parse_args(argv)
 
     app = load_config(args.config)
@@ -32,7 +34,7 @@ def main(argv: list[str] | None = None) -> int:
         scheduler.run_forever()
         return 0
     if args.command == "once":
-        print(json.dumps(scheduler.once(), indent=2, sort_keys=True))
+        print(json.dumps(scheduler.once(force=args.force), indent=2, sort_keys=True))
         return 0
     if args.command == "status":
         print(json.dumps(scheduler.status(), indent=2, sort_keys=True))

src/codex_bg/prescreen_codex.py

@@ -43,6 +43,7 @@ def screen(self, context: PreScreenContext, request: ScreeningRequest) -> Screen
                 "exec",
                 "--sandbox",
                 "read-only",
+                "--skip-git-repo-check",
                 "--json",
                 "-o",
                 str(output_file),

src/codex_bg/scheduler.py

@@ -67,22 +67,23 @@ def run_forever(self) -> None:
             for thread in threads:
                 thread.join(timeout=5)
 
-    def once(self) -> dict[str, Any]:
-        self.debug("running one scheduler cycle")
-        refreshed = self.refresh_workspaces()
-        generated = self.generate_events()
+    def once(self, *, force: bool = False) -> dict[str, Any]:
+        suffix = " with forced event generation" if force else ""
+        self.debug(f"running one scheduler cycle{suffix}")
+        refreshed = self.refresh_workspaces(force=force)
+        generated = self.generate_events(force=force)
         worked = self.work_one()
         self.debug(f"cycle complete: refreshed={refreshed} generated={generated} worked={worked}")
         return {"refreshed": refreshed, "generated": generated, "worked": worked}
 
-    def refresh_workspaces(self) -> int:
+    def refresh_workspaces(self, *, force: bool = False) -> int:
         self.debug("refreshing due shared workspaces")
-        return self.workspace_manager.refresh_due_workspaces()
+        return self.workspace_manager.refresh_due_workspaces(force=force)
 
-    def generate_events(self) -> int:
+    def generate_events(self, *, force: bool = False) -> int:
         count = 0
         for loaded in self.plugins.values():
-            if not self._plugin_due(loaded.config):
+            if not force and not self._plugin_due(loaded.config):
                 self.debug(f"plugin {loaded.config.name} is not due")
                 continue
             self.debug(f"generating events with plugin {loaded.config.name}")

src/codex_bg/workspace.py

@@ -34,14 +34,14 @@ def prepare(self, task: Task) -> Path | None:
             raise WorkspaceError(f"unknown workspace_key: {task.workspace_key}")
         return self._prepare_workspace(workspace, force_refresh=True)
 
-    def refresh_due_workspaces(self) -> int:
+    def refresh_due_workspaces(self, *, force: bool = False) -> int:
         refreshed = 0
         self.app.workspace_root.mkdir(parents=True, exist_ok=True)
         for workspace in self.app.workspaces.values():
             if not workspace.repo:
                 continue
             path = self._workspace_path(workspace)
-            if not self._refresh_due(path):
+            if not force and not self._refresh_due(path):
                 self.debug(f"workspace {workspace.key} refresh not due")
                 continue
             self._prepare_workspace(workspace, force_refresh=True)

tests/test_cli.py

@@ -47,6 +47,18 @@ def test_dry_run_is_accepted_after_subcommand(self) -> None:
         app = scheduler.call_args.args[0]
         self.assertTrue(app.dry_run)
 
+    def test_force_is_accepted_for_once(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            config = Path(tmp) / "scheduler.toml"
+            config.write_text("plugins = []\n", encoding="utf-8")
+            with patch("codex_bg.cli.Scheduler") as scheduler, patch("sys.stdout", io.StringIO()):
+                scheduler.return_value.once.return_value = {"generated": 0, "worked": False}
+
+                code = cli.main(["once", "--config", str(config), "--force"])
+
+        self.assertEqual(code, 0)
+        scheduler.return_value.once.assert_called_once_with(force=True)
+
 
 if __name__ == "__main__":
     unittest.main()

tests/test_prescreen_codex.py

@@ -36,6 +36,7 @@ def test_uses_default_small_model_when_no_model_is_configured(self) -> None:
 
         self.assertTrue(result.allowed)
         self.assertIn("--model", runner.calls[0])
+        self.assertIn("--skip-git-repo-check", runner.calls[0])
         self.assertEqual(runner.calls[0][runner.calls[0].index("--model") + 1], "gpt-5.4-mini")
         self.assertIn('model_reasoning_effort="medium"', runner.calls[0])
 

tests/test_scheduler.py

@@ -161,6 +161,33 @@ def test_zero_interval_plugin_runs_every_cycle(self) -> None:
 
             self.assertEqual(plugin.generate_calls, 2)
 
+    def test_force_once_runs_plugin_even_when_not_due(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            plugin = MemoryPlugin()
+            module = types.ModuleType("test_memory_plugin_force")
+            module.create_plugin = lambda config: plugin
+            import sys
+
+            sys.modules["test_memory_plugin_force"] = module
+            app = AppConfig(
+                database_path=Path(tmp) / "state.sqlite3",
+                workdir_root=Path(tmp) / "workdirs",
+                codex=CodexConfig(),
+                plugins=[
+                    PluginConfig(
+                        name="memory",
+                        module="test_memory_plugin_force",
+                        interval_seconds=900,
+                    )
+                ],
+            )
+            scheduler = Scheduler(app, store=Store(app.database_path), runner=FakeRunner())  # type: ignore[arg-type]
+
+            scheduler.once()
+            scheduler.once(force=True)
+
+            self.assertEqual(plugin.generate_calls, 2)
+
     def test_failed_scheduled_plugin_run_records_attempt_for_backoff(self) -> None:
         with tempfile.TemporaryDirectory() as tmp:
             plugin = FailingPlugin()