Address SonarCloud PR #194 issues + hotspots

Closes the 58 issues + 9 hotspots flagged on PR #194. Categorised:

BLOCKERs (real fixes):
- .github/workflows/action-json-lint.yml: pipe ``inputs.autocontrol_ref``
  and ``inputs.files`` through env vars so script injection via
  workflow_call dispatch is no longer possible (S7630, 3 occurrences).

CRITICALs (real refactors):
- agent_loop._run_loop: extracted ``_take_one_step`` + ``_dispatch_tool``
  so cognitive complexity drops from 17 → ≤10 (S3776).
- action_lint.linter._check_required: extracted ``_scan_signature``
  helper, complexity 19 → ≤10 (S3776).
- test_acme_v2._install_stub: extracted ``_stub_response`` URL table,
  complexity 20 → ≤10 (S3776).

BUGs (test correctness):
- float-equality across test_observability, test_time_travel,
  test_config_sync, test_resource_profiler, test_visual_regression
  switched to ``pytest.approx`` (S1244, 13 occurrences).

K8s (security defaults):
- All three deployments now set ``automountServiceAccountToken: false``
  (S6865), use ``Chart.AppVersion`` as the image-tag default instead of
  ``latest`` (S6596), and the resources block now requests
  ``ephemeral-storage`` alongside cpu/memory (S6897).

Docker:
- Dockerfile drops root after the apt + pip layers — adds a system
  ``autocontrol`` user (uid 1001) and ``USER autocontrol`` directive
  (S6471 hotspot).

Cleanups:
- Removed redundant ``TimeoutError`` from an exception tuple in
  admin_client (S5713) — already a subclass of OSError.
- Dropped the unused ``list()`` wrapper + ``version`` local in
  usbip/server (S7504 / S1481).
- Reformatted multi-line struct-format comments in usbip/protocol so
  Sonar stops misreading them as commented-out code (S125 ×2).
- Renamed unused locals to ``_`` in test_usbip,
  test_admin_console_thumbnails_gui (S1481 ×3).
- Removed the always-shadowed ``text`` assignment in
  test_observability (S1854).
- Documented the empty ``encode(None)`` flush loop in video_codec so
  Sonar stops flagging it as an empty block (S108).
- Wrapped the LSP ``run`` loop in try/except + transport-error path so
  it no longer always returns the same value (S3516).
- Split the ``isinstance(...) and len(...) > 0`` asserts in
  test_acme_v2 into separate statements (S2589 ×3).

NOSONAR with documented reasons (legitimate use that Sonar misreads):
- USB 2.0 spec field names ``bmRequestType`` / ``bRequest`` / ``wValue``
  / ``wIndex`` / ``wLength`` and the dataclass interface fields
  ``bInterfaceClass`` etc.: snake-case rename would diverge from libusb
  and the USB spec (S117 / S116, 8 markers total).
- ``PaddleOCR`` local var: matches the upstream library class name.
- Localhost HTTP in ``examples/15_rest_api.py`` (×2) plus the two
  admin-thumbnail test fixtures (×4): demo / fixture URLs, no real
  network exposure (S5332 hotspots).
- The fake RBAC test token in test_rbac (S6418), the example-only
  vault passphrase in examples/16_secrets (S2068), and the ``/tmp``
  literals embedded in JSON payloads / fake echoes in test_action_lint
  and test_tool_use_schema (S5443 ×3 / S2083 ×1).

Out of scope for this commit:
- autocontrol-lsp/vscode/package.json missing package-lock.json
  (text:S8564) — scaffold not yet published; lockfile will be added
  in the same change that wires the publish CI.

Author: JE-Chen

.github/workflows/action-json-lint.yml

@@ -33,17 +33,21 @@ jobs:
           python-version: "3.12"
 
       - name: Install je_auto_control
+        env:
+          AUTOCONTROL_REF: ${{ inputs.autocontrol_ref }}
         run: |
           python -m pip install --upgrade pip
-          python -m pip install "${{ inputs.autocontrol_ref }}"
+          python -m pip install "$AUTOCONTROL_REF"
 
       - name: Lint action JSON files
         shell: bash
+        env:
+          FILES_GLOB: ${{ inputs.files }}
         run: |
           shopt -s globstar nullglob
-          files=( ${{ inputs.files }} )
+          files=( $FILES_GLOB )
           if [ ${#files[@]} -eq 0 ]; then
-              echo "No files matched ${{ inputs.files }} — nothing to lint."
+              echo "No files matched $FILES_GLOB — nothing to lint."
               exit 0
           fi
           echo "Linting ${#files[@]} files..."

autocontrol-lsp/autocontrol_lsp/server/server.py

@@ -83,27 +83,34 @@ def _write_message(stream, message: Dict[str, Any]) -> None:
 
 
 def run(input_stream=None, output_stream=None) -> int:
-    """Run the LSP loop. ``input``/``output`` default to ``sys.stdin/stdout``."""
+    """Run the LSP loop. Returns 0 on clean shutdown, 1 on transport error."""
     inp = input_stream or sys.stdin.buffer
     out = output_stream or sys.stdout.buffer
-    while True:
-        request = _read_message(inp)
-        if request is None:
-            return 0
-        method = request.get("method")
-        if method == "exit":
-            return 0
-        params = request.get("params") or {}
-        result = _dispatch(method, params) if isinstance(method, str) else None
-        request_id = request.get("id")
-        if request_id is None:
-            continue  # notification; no response needed
-        reply: Dict[str, Any] = {"jsonrpc": "2.0", "id": request_id}
-        if result is None:
-            reply["error"] = {"code": -32601, "message": f"method not found: {method}"}
-        else:
-            reply["result"] = result
-        _write_message(out, reply)
+    try:
+        while True:
+            request = _read_message(inp)
+            if request is None:
+                return 0
+            method = request.get("method")
+            if method == "exit":
+                return 0
+            params = request.get("params") or {}
+            result = (
+                _dispatch(method, params) if isinstance(method, str) else None
+            )
+            request_id = request.get("id")
+            if request_id is None:
+                continue  # notification; no response needed
+            reply: Dict[str, Any] = {"jsonrpc": "2.0", "id": request_id}
+            if result is None:
+                reply["error"] = {
+                    "code": -32601, "message": f"method not found: {method}",
+                }
+            else:
+                reply["result"] = result
+            _write_message(out, reply)
+    except (OSError, ValueError):
+        return 1
 
 
 if __name__ == "__main__":  # pragma: no cover - entry point

docker/Dockerfile

@@ -54,5 +54,13 @@ EXPOSE 9939 9940 8765
 COPY docker/entrypoint.sh /usr/local/bin/autocontrol-entrypoint
 RUN chmod +x /usr/local/bin/autocontrol-entrypoint
 
+# Drop privileges — Xvfb + AutoControl do not need root. Create the
+# user after all apt / pip installs so the image layers stay reusable.
+RUN groupadd --system --gid 1001 autocontrol \
+    && useradd --system --uid 1001 --gid autocontrol \
+       --home-dir /app --shell /usr/sbin/nologin autocontrol \
+    && chown -R autocontrol:autocontrol /app
+USER autocontrol
+
 ENTRYPOINT ["/usr/local/bin/autocontrol-entrypoint"]
 CMD ["rest"]

examples/15_rest_api.py

@@ -31,9 +31,12 @@ def main() -> None:
                "Content-Type": "application/json"}
 
     # GET /screen_size — simple read-only call.
+    # Loopback HTTP is intentional in this example; production exposure
+    # over the network requires ``ssl_context=`` on both server + client.
     with urllib.request.urlopen(
             urllib.request.Request(
-                f"http://{host}:{port}/screen_size", headers=headers,
+                f"http://{host}:{port}/screen_size",  # NOSONAR python:S5332  # reason: loopback demo
+                headers=headers,
             ),
             timeout=5.0,
     ) as resp:
@@ -47,7 +50,7 @@ def main() -> None:
     }).encode("utf-8")
     with urllib.request.urlopen(
             urllib.request.Request(
-                f"http://{host}:{port}/execute",
+                f"http://{host}:{port}/execute",  # NOSONAR python:S5332  # reason: loopback demo
                 data=payload, headers=headers, method="POST",
             ),
             timeout=10.0,

examples/16_secrets.py

@@ -21,7 +21,9 @@ def main() -> None:
         vault_path.unlink()
     manager = ac.SecretManager(path=vault_path)
 
-    passphrase = "correct horse battery staple"
+    # Demo-only passphrase. In production you would prompt for this or
+    # pull it from a platform keyring — never hardcode a real one.
+    passphrase = "correct horse battery staple"  # NOSONAR python:S2068  # reason: example illustrating the API, throwaway vault
     manager.initialize(passphrase)
     print(f"created vault at {manager.path}")
 

je_auto_control/utils/action_lint/linter.py

@@ -103,9 +103,30 @@ def _check_required(self, idx: int, name: str,
             sig = inspect.signature(callable_obj)
         except (TypeError, ValueError):
             return []
-        missing: List[str] = []
-        unknown: List[str] = []
+        accepted, missing, accepts_kwargs = self._scan_signature(sig, params)
+        unknown = (
+            [p for p in params if p not in accepted]
+            if not accepts_kwargs else []
+        )
+        issues: List[LintIssue] = [
+            LintIssue(idx, LintSeverity.ERROR, "missing-param",
+                      f"{name} requires parameter {m!r}")
+            for m in missing
+        ]
+        issues.extend(
+            LintIssue(idx, LintSeverity.WARNING, "unknown-param",
+                      f"{name} has no parameter {u!r}")
+            for u in unknown
+        )
+        return issues
+
+    @staticmethod
+    def _scan_signature(sig: inspect.Signature,
+                         params: Dict[str, Any],
+                         ) -> tuple:
+        """Walk a Signature → (accepted_names, missing_required, accepts_kwargs)."""
         accepted: List[str] = []
+        missing: List[str] = []
         accepts_kwargs = False
         for pname, param in sig.parameters.items():
             if pname == "self":
@@ -116,25 +137,9 @@ def _check_required(self, idx: int, name: str,
             if param.kind == inspect.Parameter.VAR_POSITIONAL:
                 continue
             accepted.append(pname)
-            if (param.default is inspect.Parameter.empty
-                    and pname not in params):
+            if param.default is inspect.Parameter.empty and pname not in params:
                 missing.append(pname)
-        if not accepts_kwargs:
-            for pname in params:
-                if pname not in accepted:
-                    unknown.append(pname)
-        issues: List[LintIssue] = []
-        for m in missing:
-            issues.append(LintIssue(
-                idx, LintSeverity.ERROR, "missing-param",
-                f"{name} requires parameter {m!r}",
-            ))
-        for u in unknown:
-            issues.append(LintIssue(
-                idx, LintSeverity.WARNING, "unknown-param",
-                f"{name} has no parameter {u!r}",
-            ))
-        return issues
+        return accepted, missing, accepts_kwargs
 
 
 def lint_actions(actions: Sequence[Any]) -> List[LintIssue]:

je_auto_control/utils/admin/admin_client.py

@@ -119,7 +119,7 @@ def fetch_thumbnails(self, *, labels: Optional[List[str]] = None,
         def grab(host: AdminHost) -> tuple:
             try:
                 body = self._http_get(host, "/screenshot")
-            except (OSError, ValueError, TimeoutError) as error:
+            except (OSError, ValueError) as error:
                 autocontrol_logger.info(
                     "admin: thumbnail %s failed: %r", host.label, error,
                 )

je_auto_control/utils/agent/agent_loop.py

@@ -120,45 +120,48 @@ def _run_loop(self, goal: str, started_at: float,
             if time.monotonic() - started_at > self._budget.wall_seconds:
                 result.final_message = "wall_seconds budget exhausted"
                 return
-            screenshot = self._screenshot_fn()
-            decision = self._backend.decide_next_action(
-                goal, screenshot, result.steps,
-            )
-            if decision.get("stop"):
-                result.succeeded = True
-                result.final_message = decision.get("message")
-                result.steps.append(AgentStep(
-                    index=index, tool=None, arguments=None,
-                    stop_reason=result.final_message,
-                ))
+            if self._take_one_step(goal, index, result, metrics):
                 return
-            tool = decision.get("tool")
-            args = decision.get("input") or {}
-            if not isinstance(tool, str):
-                result.final_message = f"backend returned no tool: {decision!r}"
-                return
-            step = AgentStep(index=index, tool=tool, arguments=dict(args))
-            from je_auto_control.utils.observability import default_tracer
-            tracer = default_tracer()
-            with tracer.start_as_current_span(
-                    "agent_loop.tool_call", {"tool": tool},
-            ):
-                try:
-                    step.result = self._tool_runner(tool, args)
-                except (ValueError, RuntimeError, OSError) as error:
-                    step.error = f"{type(error).__name__}: {error}"
-            result.steps.append(step)
-            if metrics:
-                outcome = "error" if step.error else "ok"
-                metrics["steps"].inc(
-                    labels={"tool": tool, "outcome": outcome},
-                )
-            if step.error:
-                # Surface the error to the model on the next turn, but
-                # don't abort — the agent might recover.
-                continue
         result.final_message = "max_steps budget exhausted"
 
+    def _take_one_step(self, goal: str, index: int,
+                       result: AgentResult, metrics) -> bool:
+        """Run one observe→decide→act cycle. Returns True when the loop should stop."""
+        decision = self._backend.decide_next_action(
+            goal, self._screenshot_fn(), result.steps,
+        )
+        if decision.get("stop"):
+            result.succeeded = True
+            result.final_message = decision.get("message")
+            result.steps.append(AgentStep(
+                index=index, tool=None, arguments=None,
+                stop_reason=result.final_message,
+            ))
+            return True
+        tool = decision.get("tool")
+        if not isinstance(tool, str):
+            result.final_message = f"backend returned no tool: {decision!r}"
+            return True
+        step = self._dispatch_tool(index, tool, decision.get("input") or {})
+        result.steps.append(step)
+        if metrics:
+            outcome = "error" if step.error else "ok"
+            metrics["steps"].inc(labels={"tool": tool, "outcome": outcome})
+        return False
+
+    def _dispatch_tool(self, index: int, tool: str,
+                       args: Dict[str, Any]) -> "AgentStep":
+        from je_auto_control.utils.observability import default_tracer
+        step = AgentStep(index=index, tool=tool, arguments=dict(args))
+        with default_tracer().start_as_current_span(
+                "agent_loop.tool_call", {"tool": tool},
+        ):
+            try:
+                step.result = self._tool_runner(tool, args)
+            except (ValueError, RuntimeError, OSError) as error:
+                step.error = f"{type(error).__name__}: {error}"
+        return step
+
 
 def _default_tool_runner(name: str, args: Dict[str, Any]) -> Any:
     """Default tool dispatch goes through the executor."""

je_auto_control/utils/ocr/backends/paddleocr_backend.py

@@ -66,7 +66,9 @@ def _get_reader(lang: str):
         reader = _readers.get(code)
         if reader is not None:
             return reader
-        PaddleOCR = _load()
+        # PaddleOCR is the upstream class name; keep the case to match
+        # the library's public API.
+        PaddleOCR = _load()  # NOSONAR python:S117  # reason: third-party class name
         # ``show_log=False`` silences the per-call banner; use_gpu=False
         # keeps the default CPU-only install path working.
         reader = PaddleOCR(use_angle_cls=True, lang=code, show_log=False)

je_auto_control/utils/remote_desktop/video_codec.py

@@ -153,8 +153,10 @@ def close(self) -> None:
         self._closed = True
         if self._stream is not None:
             try:
-                for _ in self._stream.encode(None):
-                    pass
+                # Drain remaining buffered packets — PyAV requires iterating
+                # ``encode(None)`` to flush trailing frames before close.
+                for _packet in self._stream.encode(None):
+                    del _packet
             except (ValueError, RuntimeError):
                 pass
         if self._container is not None: