fix(models,cli,batch): free-VRAM sizing + device honesty, gguf seed, offline errors

Author: ctrl-gaurav

effgen/cli/_main.py

@@ -2187,14 +2187,59 @@ def _models_info(self, args):
             return 1
 
         from effgen.models import _catalog, _refresh
+        from effgen.models.model_loader import ModelLoader
+
+        # An engine-prefixed id (e.g. "transformers:Qwen/Qwen2.5-1.5B-Instruct")
+        # names a local engine + a bare repo id. Strip the prefix so the id
+        # matches the local cache and the catalog, and remember the engine so we
+        # can lead with the local view.
+        lookup_name = args.name
+        requested_engine = None
+        if ":" in lookup_name:
+            _prefix, _rest = lookup_name.split(":", 1)
+            if _prefix in ModelLoader._LOCAL_ENGINE_PREFIXES and _rest:
+                requested_engine = _prefix
+                lookup_name = _rest
 
         # Is this id sitting in the local HF cache? If so we can describe it as
         # locally-runnable even when the cloud catalog has no (or a different) row.
         local_entry = next(
-            (m for m in self._local_cached_models() if m["id"] == args.name), None
+            (m for m in self._local_cached_models() if m["id"] == lookup_name), None
         )
 
-        rec = _catalog.lookup(args.name)
+        # An explicit local-engine request is answered from the local cache: show
+        # the cached copy, or report a cache miss naming what is cached (rather
+        # than cloud catalog suggestions the engine can't run).
+        if requested_engine is not None:
+            if local_entry is not None:
+                local_payload = self._local_model_payload(local_entry)
+                if getattr(args, "output_json", False):
+                    print(json.dumps({"id": lookup_name, "provider": None,
+                                       "engine": requested_engine,
+                                       "local": local_payload}, indent=2))
+                    return 0
+                self.print_header(f"Model: {lookup_name} ({requested_engine})")
+                self._render_local_model_info(local_payload)
+                return 0
+            cached = [m["id"] for m in self._local_cached_models()]
+            if getattr(args, "output_json", False):
+                print(json.dumps({"id": lookup_name, "provider": None,
+                                   "engine": requested_engine, "local": None,
+                                   "cached_models": cached}, indent=2))
+                return 1
+            self.print_error(
+                f"Model '{lookup_name}' is not in the local cache, so the "
+                f"'{requested_engine}' engine can't run it yet."
+            )
+            if cached:
+                self.print("Locally cached models:")
+                for cid in cached:
+                    self.print(f"  {cid}")
+                self.print(f"\nDownload it first: effgen run -m {lookup_name} "
+                           f"--engine {requested_engine} \"...\" (with network access).")
+            return 1
+
+        rec = _catalog.lookup(lookup_name)
         if rec is None:
             if local_entry is not None:
                 # Downloaded locally but not in the cloud catalog: describe the local
@@ -2318,6 +2363,9 @@ def _models_unload(self, args):
 
     def _models_status(self, args):
         """Show loaded models and GPU memory status."""
+        if getattr(args, "output_json", False):
+            return self._models_status_json()
+
         self.print_header("Model & GPU Status")
 
         # GPU memory info — physical (driver) view across all processes, so this
@@ -2385,6 +2433,47 @@ def _models_status(self, args):
         registered = list_registered_models()
         self.print(f"\nCapability profiles registered: {len(registered)}")
 
+    def _models_status_json(self) -> int:
+        """Emit the GPU table + loaded models as JSON for ops/edge tooling."""
+        gib = 1024 ** 3
+        gpu_list: list[dict] = []
+        cuda_available = True
+        try:
+            from effgen.gpu.cuda_compat import per_gpu_status
+            for g in per_gpu_status():
+                gpu_list.append({
+                    "index": g.index,
+                    "name": g.name,
+                    "total_gb": round(g.total_bytes / gib, 3),
+                    "used_gb": round(g.used_bytes / gib, 3),
+                    "free_gb": round(g.free_bytes / gib, 3),
+                    "utilization_pct": g.utilization_pct,
+                })
+            if not gpu_list:
+                try:
+                    import torch
+                    cuda_available = torch.cuda.is_available()
+                except ImportError:
+                    cuda_available = False
+        except ImportError:
+            cuda_available = False
+
+        from effgen.models.capabilities import list_registered_models
+        from effgen.models.model_loader import ModelLoader
+        loaded = ModelLoader().get_loaded_models()
+        loaded_list = [
+            {"name": name, "loaded": bool(model.is_loaded())}
+            for name, model in loaded.items()
+        ]
+        payload = {
+            "cuda_available": cuda_available,
+            "gpus": gpu_list,
+            "loaded_models": loaded_list,
+            "capability_profiles": len(list_registered_models()),
+        }
+        print(json.dumps(payload, indent=2))
+        return 0
+
     def _models_refresh(self, args):
         """Refresh the bundled model catalog from each provider's live API.
 
@@ -2935,7 +3024,8 @@ def create_parser():
     models_unload = models_subparsers.add_parser('unload', help='Unload a model from memory')
     models_unload.add_argument('name', help='Model name')
 
-    models_subparsers.add_parser('status', help='Show loaded models and GPU memory status')
+    models_status = models_subparsers.add_parser('status', help='Show loaded models and GPU memory status')
+    models_status.add_argument('--json', dest='output_json', action='store_true', help='Output as JSON')
 
     models_refresh = models_subparsers.add_parser(
         'refresh', help="Refresh the model catalog from each provider's live API")
@@ -4060,7 +4150,7 @@ def _read_done_indices(output_path: Path) -> dict:
 
 def _handle_batch_command(args, cli) -> int:
     """Handle the 'batch' CLI subcommand."""
-    from effgen.core.batch import BatchConfig, BatchRunner
+    from effgen.core.batch import _QUERY_ALIASES, BatchConfig, BatchRunner
 
     input_path = args.input
     output_path = getattr(args, 'output', None)
@@ -4149,14 +4239,19 @@ def _json_error(exc: Exception) -> int:
         # naming the file and line number (not the parser's byte offset);
         # --strict turns the first bad line into a hard failure instead.
         skipped: list[int] = []
+        empty_rows: list[tuple[int, list[str]]] = []
 
         def _on_skip(lineno: int, msg: str) -> None:
             skipped.append(lineno)
             cli.print(f"Skipping malformed input at {input_path}:{lineno}: {msg}")
 
+        def _on_empty(lineno: int, keys: list[str]) -> None:
+            empty_rows.append((lineno, keys))
+
         try:
             queries = runner._read_queries(
-                Path(input_path), query_field, strict=strict, on_skip=_on_skip,
+                Path(input_path), query_field, strict=strict,
+                on_skip=_on_skip, on_empty=_on_empty,
             )
         except Exception as e:  # noqa: BLE001 - one clear message, no traceback
             cli.print_error(f"Could not read {input_path}: {e}")
@@ -4167,6 +4262,22 @@ def _on_skip(lineno: int, msg: str) -> None:
                 f"{len(queries)} queries loaded."
             )
 
+        # A row with no recognized query text (neither --query-field nor the
+        # aliases query/input/prompt/question/text) can't run. Name the fields it
+        # did carry and how to point at the right one, rather than letting each
+        # empty row fail with a generic empty-task message.
+        if empty_rows:
+            lineno, keys = empty_rows[0]
+            fields = ", ".join(keys) if keys else "none"
+            more = f" (and {len(empty_rows) - 1} more)" if len(empty_rows) > 1 else ""
+            msg = (
+                f"Row {lineno}{more} has no query text. Fields present: {fields}. "
+                f"Set the query column with --query-field NAME, or key rows on one "
+                f"of: {', '.join(_QUERY_ALIASES)}."
+            )
+            cli.print_error(msg)
+            return _json_error(ValueError(msg))
+
         # --resume: skip input rows already present in the JSONL output.
         done_rows: dict[int, dict] = {}
         if resume:

effgen/core/agent.py

@@ -619,7 +619,14 @@ def __init__(self, config: AgentConfig | None = None, session_id: str | None = N
                     # Mark closed first so this never-returned, partially-built
                     # agent doesn't tail the error with a GC-without-close warning.
                     self._closed = True
-                    logger.error(f"Failed to load model '{self.model_name}': {e}")
+                    # A typed load error (e.g. an offline cache miss or a
+                    # require_gpu policy failure) already carries a clear,
+                    # user-facing message and is re-raised below — log it at
+                    # debug so it isn't repeated as an ERROR alongside the raise.
+                    if type(e).__name__ in ("ModelNotCachedError", "GPUPlacementError"):
+                        logger.debug(f"Failed to load model '{self.model_name}': {e}")
+                    else:
+                        logger.error(f"Failed to load model '{self.model_name}': {e}")
                     raise RuntimeError(
                         f"Failed to load model '{self.model_name}': {e}"
                     ) from e

effgen/core/agent_generation.py

@@ -586,6 +586,12 @@ def _accumulate_run_cost(self, result_metadata: dict[str, Any] | None) -> None:
             val = result_metadata.get(key)
             if isinstance(val, int | float):
                 accum[key] = accum.get(key, 0) + val
+        # Where the model ran (local engines report 'cuda'/'cpu'/'mixed'), so a
+        # headless caller can detect a CPU fallback. Not summed — the last call
+        # wins, and cloud calls that don't report it leave it untouched.
+        device = result_metadata.get("device")
+        if device:
+            accum["device"] = device
         accum["calls"] = accum.get("calls", 0) + 1
 
     def _finalize_cost_metadata(self, response: AgentResponse) -> None:
@@ -612,6 +618,8 @@ def _finalize_cost_metadata(self, response: AgentResponse) -> None:
         for key in ("prompt_tokens", "completion_tokens", "total_tokens"):
             if key in accum and key not in meta:
                 meta[key] = int(accum[key])
+        if accum.get("device") and "device" not in meta:
+            meta["device"] = accum["device"]
         # response.tokens_used is documented as the run's total token count;
         # the value set along the generation path is completion-tokens-only
         # (each provider adapter's GenerationResult.tokens_used). Correct it

effgen/core/batch.py

@@ -29,6 +29,26 @@
 
 logger = logging.getLogger(__name__)
 
+# Field names accepted for the query text of a JSONL/CSV/JSON row, in priority
+# order — mirrors the eval/compare loaders so a batch file keyed on any of these
+# works without ``--query-field``.
+_QUERY_ALIASES = ("query", "input", "prompt", "question", "text")
+
+
+def _resolve_row_query(obj: dict[str, Any], query_field: str) -> str:
+    """Return a row's query text, trying *query_field* then the common aliases.
+
+    The explicitly-requested field wins; when it is absent or empty the aliases
+    ``input``/``prompt``/``question``/``text`` are tried in order. Returns an
+    empty string when none carry text.
+    """
+    order = [query_field] + [k for k in _QUERY_ALIASES if k != query_field]
+    for key in order:
+        val = obj.get(key)
+        if val not in (None, ""):
+            return str(val)
+    return ""
+
 
 @dataclass
 class BatchConfig:
@@ -496,9 +516,15 @@ def _read_queries(
         query_field: str,
         strict: bool = False,
         on_skip: Callable[[int, str], None] | None = None,
+        on_empty: Callable[[int, list[str]], None] | None = None,
     ) -> list[str]:
         """Read queries from a JSONL, CSV, JSON, or plain-text file.
 
+        A dict row's query text is read from *query_field*, falling back to the
+        common aliases ``input``/``prompt``/``question``/``text`` when that field
+        is absent — so a file keyed on any of those works without
+        ``--query-field``.
+
         A malformed JSONL/JSON line does not abort the whole job. By default the
         bad line is skipped and reported (through ``on_skip`` if given, else a
         logged warning) with the **file path and line number** — not the JSON
@@ -511,6 +537,9 @@ def _read_queries(
             strict: Hard-fail on the first malformed line instead of skipping.
             on_skip: Called with ``(line_number, message)`` for each skipped
                 line; when ``None``, a warning is logged instead.
+            on_empty: Called with ``(line_number, available_keys)`` for each dict
+                row that carries no recognized query text, so the caller can name
+                the fields it saw.
         """
         suffix = path.suffix.lower()
         queries: list[str] = []
@@ -524,6 +553,12 @@ def _report_bad(lineno: int, exc: Exception) -> None:
             else:
                 logger.warning("%s: skipping malformed line: %s", location, exc)
 
+        def _resolve(lineno: int, row: dict) -> str:
+            text = _resolve_row_query(row, query_field)
+            if not text and on_empty is not None:
+                on_empty(lineno, sorted(str(k) for k in row.keys()))
+            return text
+
         if suffix == ".jsonl":
             with open(path, encoding="utf-8") as f:
                 for lineno, line in enumerate(f, start=1):
@@ -538,14 +573,14 @@ def _report_bad(lineno: int, exc: Exception) -> None:
                     if isinstance(obj, str):
                         queries.append(obj)
                     elif isinstance(obj, dict):
-                        queries.append(str(obj.get(query_field, "")))
+                        queries.append(_resolve(lineno, obj))
                     else:
                         queries.append(str(obj))
         elif suffix == ".csv":
             with open(path, encoding="utf-8") as f:
                 reader = csv.DictReader(f)
-                for row in reader:
-                    queries.append(row.get(query_field, ""))
+                for lineno, row in enumerate(reader, start=1):
+                    queries.append(_resolve(lineno, row))
         elif suffix == ".json":
             with open(path, encoding="utf-8") as f:
                 try:
@@ -557,7 +592,7 @@ def _report_bad(lineno: int, exc: Exception) -> None:
                         if isinstance(item, str):
                             queries.append(item)
                         elif isinstance(item, dict):
-                            queries.append(str(item.get(query_field, "")))
+                            queries.append(_resolve(pos, item))
                         else:
                             _report_bad(pos, TypeError(
                                 f"expected string or object, got {type(item).__name__}"

effgen/models/gguf_engine.py

@@ -114,6 +114,10 @@ def generate(
             self.load()
         params = self._to_kwargs(config)
         params.update(kwargs)
+        # Clear residual context/KV state from any prior call so each completion
+        # starts from a clean state. Without this the sampler's RNG continues
+        # from the previous call and a fixed seed does not reproduce.
+        self._llm.reset()
         out = self._llm(prompt, **params)
         choice = out["choices"][0]
         text = choice.get("text", "")
@@ -137,6 +141,7 @@ def generate_stream(
         params = self._to_kwargs(config)
         params.update(kwargs)
         params["stream"] = True
+        self._llm.reset()
         for chunk in self._llm(prompt, **params):
             yield chunk["choices"][0].get("text", "")