feat: rjob rl with cyber env

.gitignore

@@ -30,3 +30,11 @@ tmp/
 
 # Local Harbor dataset tooling
 /env/harbor/generate_vulhub_dataset.py
+
+# Local cybergym env overrides (tracked files kept via skip-worktree)
+/env/cybergym/
+
+/.safactory-locks/
+
+# Local credentials / private config (do not commit)
+config.yaml

args.py

@@ -138,7 +138,14 @@ def parse_simulation_args(argv: Sequence[str] | None = None) -> argparse.Namespa
         "--gateway-close-timeout-s",
         type=float,
         default=120.0,
-        help="Total timeout for polling gateway session close completion.",
+        help=(
+            "HTTP timeout for gateway session close requests. Must exceed the"
+            " gateway's drain_timeout_s (default 30s): the close endpoint blocks"
+            " up to drain_timeout_s waiting for in-flight LLM requests to finish,"
+            " so a runner timeout shorter than drain_timeout_s abandons the close"
+            " before the gateway responds, leaving the session unsealed"
+            " (is_terminal=0) and orphaning the rollout group."
+        ),
     )
     parser.add_argument(
         "--gateway-close-retries",

clusters/rjob_cluster.py

@@ -24,6 +24,9 @@
 
 _DEFAULT_RUNNER_CONTAINER_PATH = "/tmp/safactory-openclaw-runner.mjs"
 _DEFAULT_RUN_COMMAND = f"node {_DEFAULT_RUNNER_CONTAINER_PATH}"
+# RJob task ids are "<job_name>-<task_name>" and must match the cluster regex
+# ^[a-zA-Z0-9][-a-zA-Z0-9]{1,61}[a-zA-Z0-9]$ — dots are NOT allowed, so strip them
+# (replace with "-") rather than only allowing alnum + "." + "-".
 _INVALID_NAME_CHARS = re.compile(r"[^a-z0-9-]+")
 _MAX_RJOB_NAME_LEN = 49
 _MAX_RJOB_AGENT_NAME_LEN = 12
@@ -279,6 +282,10 @@ async def wait_terminal(
                     trace.update_context(rjob_submit_to_starting_ms=submit_to_starting_ms)
             if status == "Running" and submit_to_running_ms is None:
                 submit_to_running_ms = elapsed_ms
+                # Absolute epoch seconds at the moment the RJob entered Running.
+                # Joined with rjob_submit_ts to derive cluster queue time
+                # (rjob_running_ts - rjob_submit_ts) in the episode record.
+                rjob_running_ts = time.time()
                 if trace is not None:
                     trace.mark(
                         "rjob_running",
@@ -287,7 +294,10 @@ async def wait_terminal(
                         job_name=job_name,
                         submit_to_running_ms=submit_to_running_ms,
                     )
-                    trace.update_context(rjob_submit_to_running_ms=submit_to_running_ms)
+                    trace.update_context(
+                        rjob_submit_to_running_ms=submit_to_running_ms,
+                        rjob_running_ts=rjob_running_ts,
+                    )
             if status != last_status:
                 if trace is not None:
                     trace.mark(

config.yaml.example

@@ -0,0 +1,33 @@
+# Global RJob connection/auth settings shared by all RJob environments.
+# Override with: python launcher.py --mode rjob --rjob-config path/to/config.yaml
+#
+# Copy this file to config.yaml and fill in your own credentials.
+# Do NOT commit config.yaml (it is gitignored).
+
+rjob:
+  cluster_entry: "https://h.pjlab.org.cn"
+  namespace: "ailab-evobox"
+  access_key: "<YOUR_ACCESS_KEY>"
+  secret_key: "<YOUR_SECRET_KEY>"
+  verifyssl: true
+  retries: 3
+
+  # Optional but commonly shared across RJob submissions.
+  charged_group: "evobox_proxy"
+  # NOTE: intentionally NOT set. rjob_cluster.resolve_gateway_base_url does
+  # `cfg.get("gateway_base_url") or request.gateway_base_url`; if we hardcode an
+  # IP here it goes stale the moment the training pod restarts (the runner then
+  # gets a dead gateway URL and LLM calls hang). Leave this unset so it falls
+  # back to request.gateway_base_url, which the buffer server fills from
+  # AIEVOBOX_GATEWAY_BASE_URL (env.rjob.sh: hostname -I of THIS pod).
+  # gateway_base_url: "http://<gateway_ip>:8000/v1/sessions"
+  name_prefix: safactory
+  poll_interval_s: 5
+  # Set to false temporarily so succeeded-but-empty pods are kept for log inspection.
+  cleanup_on_finish: false
+  keep_failed_jobs: true
+  no_packaging: true
+  auto_delete_duration: 12h
+  # Raised from 1 so multiple rollout episodes can be submitted in parallel
+  # (must be >= AIEVOBOX_POOL_SIZE to actually parallelize).
+  submit_concurrency: 8

core/data_manager/manager.py

@@ -314,6 +314,21 @@ async def list_session_steps(
             checkout_latest=checkout_latest,
         ))
 
+    async def list_terminal_steps_for_sessions(
+        self,
+        session_ids: List[str],
+        *,
+        job_id: Optional[str] = None,
+    ) -> List[Dict[str, Any]]:
+        """Return terminal step rows for a batch of sessions (used by the RL buffer)."""
+        if not session_ids:
+            return []
+        return await self._strategy.list_session_step_rows(SessionStepQuery(
+            job_id=job_id or self.job_id or None,
+            session_ids=tuple(session_ids),
+            is_terminal=True,
+        ))
+
     async def update_session_step_rows(
         self,
         *,
@@ -418,24 +433,7 @@ async def mark_latest_session_completed(
     async def close(self) -> None:
         """Close the storage strategy"""
         await self._strategy.close()
-
-    async def fetch_done_steps_with_context(
-        self,
-        after_id: int = 0,
-        limit: int = 100
-    ) -> List[Dict]:
-        """Fetch completed steps for training data collection"""
-        if hasattr(self._strategy, 'fetch_done_steps_with_context'):
-            return await self._strategy.fetch_done_steps_with_context(self.job_id, after_id, limit)
-        return []
-
-    async def get_max_step_id(self) -> int:
-        """Get maximum primary key for pagination"""
-        if hasattr(self._strategy, 'get_max_step_id'):
-            return await self._strategy.get_max_step_id(self.job_id)
-        return 0
 
-    @property
     def buffer_stats(self) -> Optional[dict]:
         """Get buffer statistics (SQLite only)"""
         if hasattr(self._strategy, 'buffer_stats'):

core/data_manager/strategy/cloud_strategy_impl.py

@@ -1133,78 +1133,6 @@ async def _flush_records(self) -> int:
             self._stats["flush_count"] += 1
             log.debug("Flushed %d cloud records", len(records))
             return len(records)
-
-    async def fetch_done_steps_with_context(
-        self,
-        job_id: str,
-        after_id: int = 0,
-        limit: int = 100
-    ) -> List[Dict]:
-        """
-        Fetch completed steps for training data collection.
-        Uses cursor-based pagination.
-        """
-        await self.init()
-
-        results = self.client.pull_data(
-            dataset_type=CLOUD_DATASET_TYPE,
-            cursor=after_id,
-            checkout_latest=True,
-            where_sql="job_id = '{}' AND is_terminal = True".format(_escape_sql_literal(job_id)),
-            limit=limit,
-            deserialize_json=True,
-        )
-
-        if results is None or len(results) == 0:
-            log.debug("No completed cloud steps to fetch: result_count=%s", 0 if results is None else len(results))
-            return []
-
-        cursor = self.client.extract_cursor(results)
-
-        rows = []
-        for _, row in results.iterrows():
-            meta = _meta_json_object(row.get("meta_json"))
-            messages = _json_value(row.get("messages"), [])
-            if not isinstance(messages, (dict, list)):
-                messages = []
-            response = _json_value(
-                row.get("response"),
-                row.get("response"),
-            )
-            rows.append(
-                {
-                    "step_pk": cursor,
-                    "step_id": row["step_id"],
-                    "env_name": row["env_name"],
-                    "env_id": row["session_id"],
-                    "meta_json": json.dumps(meta, ensure_ascii=False, default=str),
-                    "prompt": self.normalize_messages(messages),
-                    "request": meta.get("request"),
-                    "response": _response_text(response),
-                    "reward": row["reward"],
-                    "step_reward": row["step_reward"],
-                    "total_reward": row["reward"],
-                    "session_id": row["session_id"],
-                    "session_end_time": row["created_at"] if row["created_at"] else None,
-                    "group_id": meta.get("group_id"),
-                    "truncated": row["is_truncated"],
-                    "is_session_completed": row["is_session_completed"], 
-                }
-            )
-        return rows
-
-    async def get_max_step_id(self, job_id: str) -> int:
-        """Get maximum primary key for pagination"""
-        await self.init()
-
-        last_cursor = self.client.get_max_created_at(
-            where_sql=(
-                "dataset_type = '{}' AND job_id = '{}' AND is_terminal = True"
-                .format(CLOUD_DATASET_TYPE, _escape_sql_literal(job_id))
-            ),
-        )
-
-        return last_cursor
 
     # --- Helpers ---
     def extract_image_path(self, item: dict) -> str | None:

core/data_manager/strategy/sqlite_strategy_impl.py

@@ -536,82 +536,3 @@ def buffer_stats(self) -> Optional[dict]:
         """Get buffer statistics"""
         return self._write_buffer.stats if self._write_buffer else None
 
-    async def fetch_done_steps_with_context(
-        self,
-        job_id: str,
-        after_id: int = 0,
-        limit: int = 100
-    ) -> List[Dict]:
-        """
-        Fetch completed steps for training data collection.
-        Uses cursor-based pagination.
-        """
-        await self.init()
-
-        trace = PerfTrace(
-            "sqlite_strategy.fetch_done_steps_with_context",
-            logger=log,
-            context={
-                "operation": "db_read",
-                "table": "session_steps",
-                "job_id": job_id,
-                "after_id": after_id,
-                "limit": limit,
-            },
-        )
-        try:
-            with trace.span("db_read.fetch_done_steps", limit=limit):
-                steps = await SessionStep.filter(
-                    job_id=job_id,
-                    is_trainable=True,
-                    id__gt=after_id
-                ).order_by("id").limit(limit)
-
-            rows = [
-                {
-                    "step_pk": s.id,
-                    "step_id": s.step_id,
-                    "env_name": s.env_name,
-                    "env_id": s.session_id,
-                    "meta_json": s.meta_json,
-                    "prompt": s.messages,
-                    "request": s.request,
-                    "response": s.response,
-                    "reward": s.step_reward,
-                    "step_reward": s.step_reward,
-                    "total_reward": s.reward,
-                    "session_id": s.session_id,
-                    "session_end_time": s.created_at.isoformat() if s.created_at else None,
-                    "group_id": s.group_id,
-                    "truncated": s.is_truncated,
-                    "is_session_completed": s.is_session_completed,
-                }
-                for s in steps
-            ]
-            trace.emit_summary(status="success", row_count=len(rows))
-            return rows
-        except Exception as exc:
-            trace.emit_summary(status="failed", error_type=type(exc).__name__, error=str(exc))
-            raise
-
-    async def get_max_step_id(self, job_id: str) -> int:
-        """Get maximum primary key for pagination"""
-        await self.init()
-        trace = PerfTrace(
-            "sqlite_strategy.get_max_step_id",
-            logger=log,
-            context={
-                "operation": "db_read",
-                "table": "session_steps",
-                "job_id": job_id,
-            },
-        )
-        try:
-            with trace.span("db_read.max_terminal_step_id"):
-                latest = await SessionStep.filter(job_id=job_id, is_terminal=True).order_by("-id").first()
-            max_id = latest.id if latest else 0
-            trace.emit_summary(status="success", row_count=1 if latest else 0, max_step_id=max_id)
-            return max_id
-        except Exception as exc:
-            trace.emit_summary(status="failed", error_type=type(exc).__name__, error=str(exc))
-            raise