Show engine control in cockpit

src/braiins_ratchet/cli.py

@@ -169,7 +169,14 @@ def cmd_next(_: argparse.Namespace) -> int:
     with connect() as conn:
         init_db(conn)
         recover_stale_active_watch(conn)
-        print(build_operator_cockpit(conn))
+        engine_status = get_engine_status()
+        print(
+            build_operator_cockpit(
+                conn,
+                engine_running=engine_status.running,
+                engine_detail=engine_status.detail,
+            )
+        )
     return 0
 
 
@@ -180,13 +187,18 @@ def cmd_app_state(_: argparse.Namespace) -> int:
         recover_stale_active_watch(conn)
         operator_state = get_operator_state(conn)
         automation_plan = build_automation_plan(conn)
+        engine_status = get_engine_status()
         payload = {
             "generated_at": datetime.now(UTC).isoformat(timespec="seconds"),
             "operator_state": asdict(operator_state),
             "automation_plan": asdict(automation_plan),
-            "engine_status": asdict(get_engine_status()),
+            "engine_status": asdict(engine_status),
             "config": asdict(config),
-            "cockpit": build_operator_cockpit(conn),
+            "cockpit": build_operator_cockpit(
+                conn,
+                engine_running=engine_status.running,
+                engine_detail=engine_status.detail,
+            ),
             "latest": {
                 "ocean": _object_dict(latest_ocean_snapshot(conn)),
                 "market": _object_dict(latest_market_snapshot(conn)),

src/braiins_ratchet/guidance.py

@@ -84,7 +84,12 @@ def get_operator_state(conn) -> OperatorState:
     )
 
 
-def build_operator_cockpit(conn) -> str:
+def build_operator_cockpit(
+    conn,
+    *,
+    engine_running: bool = False,
+    engine_detail: str | None = None,
+) -> str:
     state = get_operator_state(conn)
 
     lines = [
@@ -100,6 +105,7 @@ def build_operator_cockpit(conn) -> str:
         f"  Experiment ledger: {EXPERIMENT_LOG.relative_to(REPORTS_DIR.parent) if EXPERIMENT_LOG.exists() else 'none yet'}",
         f"  Active watch: {state.active_watch or 'none detected'}",
         f"  Active manual exposure: {_manual_exposure_text(state.active_manual_positions)}",
+        f"  Forever engine: {_engine_text(engine_running, engine_detail)}",
         f"  Research stage: {_research_stage(state.active_watch, state.completed_watch)}",
     ]
 
@@ -122,6 +128,7 @@ def build_operator_cockpit(conn) -> str:
             has_market=state.has_market,
             is_fresh=state.is_fresh,
             action=state.action,
+            engine_running=engine_running,
         )
     )
     lines.extend(["", "Ratchet Pathway Forecast"])
@@ -134,6 +141,7 @@ def build_operator_cockpit(conn) -> str:
             has_market=state.has_market,
             is_fresh=state.is_fresh,
             action=state.action,
+            engine_running=engine_running,
         )
     )
     lines.extend(["", "How To Interpret The Current Action"])
@@ -169,6 +177,7 @@ def _do_this_now(
     has_market: bool,
     is_fresh: bool,
     action: str | None,
+    engine_running: bool = False,
 ) -> list[str]:
     if active_watch:
         return [
@@ -189,6 +198,14 @@ def _do_this_now(
             "    ./scripts/ratchet position close POSITION_ID",
         ]
 
+    if engine_running:
+        return [
+            "  DO NOTHING.",
+            "  The forever engine is running and owns passive research.",
+            "  It will wait through cooldown, start the next passive watch when allowed, write evidence, then repeat.",
+            "  Do not start another watch or one-shot command unless you intentionally stop the engine first.",
+        ]
+
     if completed_watch and action == "manual_canary":
         return [
             "  STOP.",
@@ -251,6 +268,7 @@ def _pathway_forecast(
     has_market: bool,
     is_fresh: bool,
     action: str | None,
+    engine_running: bool = False,
 ) -> list[str]:
     if active_watch:
         return [
@@ -268,6 +286,14 @@ def _pathway_forecast(
             "  Longterm, possible: resume passive ratchet experiments only after exposure is closed.",
         ]
 
+    if engine_running:
+        return [
+            "  Planning probabilities are workflow estimates, not profit probabilities.",
+            "  Immediate, certain: workload is zero; the engine owns the next wait/watch/report cycle.",
+            "  Midterm, likely: the app will show cooldown or live watch progress without terminal babysitting.",
+            "  Longterm, likely: the engine keeps repeating passive research stages until you stop it.",
+        ]
+
     if completed_watch and action == "manual_canary":
         return [
             "  Planning probabilities are workflow estimates, not profit probabilities.",
@@ -375,6 +401,12 @@ def _manual_exposure_text(positions: list[str]) -> str:
     return "; ".join(positions)
 
 
+def _engine_text(engine_running: bool, engine_detail: str | None) -> str:
+    if engine_running:
+        return engine_detail or "running"
+    return engine_detail or "not running"
+
+
 def _recent_completed_watch(latest_report: str | None, latest_market_timestamp: str | None) -> CompletedWatch | None:
     if latest_report is None:
         return None

src/braiins_ratchet/lifecycle.py

@@ -187,7 +187,13 @@ def run_supervisor(config: AppConfig, *, once: bool = False) -> int:
                 "watch_completed",
                 {"run_id": run_id, "next_action_utc": next_action.isoformat(timespec="seconds")},
             )
-            print(build_operator_cockpit(conn))
+            print(
+                build_operator_cockpit(
+                    conn,
+                    engine_running=True,
+                    engine_detail="forever monitor engine is running inside this supervisor process",
+                )
+            )
         if once:
             return 0
 

tests/test_guidance.py

@@ -109,6 +109,24 @@ class GuidanceTests(unittest.TestCase):
         self.assertIn("Do not start another identical watch now.", text)
         self.assertIn("./scripts/ratchet once", text)
 
+    def test_running_engine_owns_passive_research(self) -> None:
+        lines = _do_this_now(
+            active_watch=None,
+            active_manual_positions=[],
+            completed_watch=_completed_watch(age_minutes=4),
+            has_ocean=True,
+            has_market=True,
+            is_fresh=True,
+            action="manual_canary",
+            engine_running=True,
+        )
+
+        text = "\n".join(lines)
+
+        self.assertIn("DO NOTHING.", text)
+        self.assertIn("forever engine is running", text)
+        self.assertIn("wait through cooldown", text)
+
     def test_recent_completed_watch_forecast_enters_cooldown(self) -> None:
         lines = _pathway_forecast(
             active_watch=None,