fix(llm,storage): harden Windows extraction paths

reflexio/server/llm/providers/claude_code_provider.py

@@ -62,14 +62,22 @@
 _ENV_MODEL = "CLAUDE_SMART_CLI_MODEL"
 _HOST_CODEX = "codex"
 _HOST_CLAUDE_CODE = "claude-code"
+
+
+def _is_windows() -> bool:
+    return os.name == "nt"
+
+
 _CODEX_COMPAT_SCRIPT_NAMES = (
     ("codex-claude-compat.cmd", "codex-claude-compat")
-    if os.name == "nt"
+    if _is_windows()
     else ("codex-claude-compat", "codex-claude-compat.cmd")
 )
 _CODEX_COMPAT_SCRIPT_NAME_SET = set(_CODEX_COMPAT_SCRIPT_NAMES)
 _DEFAULT_TIMEOUT_SECONDS = 120
 _DEFAULT_CLI_MODEL = "claude-sonnet-5"
+_WINDOWS_ARGV_SYSTEM_PROMPT_LIMIT = 3_000
+_WINDOWS_CLI_SUFFIXES = (".cmd", ".exe", ".bat", ".ps1")
 
 _TRUTHY_ENV_VALUES = {"1", "true", "yes"}
 _UNSUPPORTED_PARAMS_WARNED: set[str] = set()
@@ -116,6 +124,39 @@ def _candidate_codex_compat_path() -> Path | None:
     return None
 
 
+def _windows_cli_suffixes() -> tuple[str, ...]:
+    suffixes: list[str] = []
+    for suffix in os.environ.get("PATHEXT", "").split(";"):
+        suffix = suffix.strip().lower()
+        if not suffix:
+            continue
+        suffixes.append(suffix if suffix.startswith(".") else f".{suffix}")
+    for suffix in _WINDOWS_CLI_SUFFIXES:
+        if suffix not in suffixes:
+            suffixes.append(suffix)
+    return tuple(suffixes)
+
+
+def _resolve_cli_override_path(cli_path: str) -> str:
+    if not _is_windows():
+        return cli_path
+    path = Path(cli_path)
+    if path.suffix:
+        return cli_path
+
+    # Windows package managers expose extensioned shims while users often
+    # configure the extensionless binary name.
+    for suffix in _windows_cli_suffixes():
+        candidate = path.with_suffix(suffix)
+        if candidate.exists():
+            return str(candidate)
+    return cli_path
+
+
+def _subprocess_text_errors() -> str:
+    return "replace" if _is_windows() else "strict"
+
+
 def _resolve_cli_path() -> str | None:
     """Return the path to the active host CLI, or None if unavailable.
 
@@ -129,7 +170,7 @@ def _resolve_cli_path() -> str | None:
     """
     override = os.environ.get(_ENV_CLI_PATH)
     if override:
-        candidate = Path(override)
+        candidate = Path(_resolve_cli_override_path(override))
         if candidate.is_file() and os.access(candidate, os.X_OK):
             return str(candidate)
         _LOGGER.warning(
@@ -432,6 +473,7 @@ def _run_claude_stream(
 ) -> ParseResult:
     """Invoke ``claude -p --output-format stream-json`` and return a ParseResult."""
     model = os.environ.get(_ENV_MODEL) or _DEFAULT_CLI_MODEL
+    inline_system_prompt = _should_inline_system_prompt(system_prompt)
     cmd = [
         cli_path,
         "-p",
@@ -442,8 +484,13 @@ def _run_claude_stream(
         "--model",
         model,
     ]
-    if system_prompt:
+    if system_prompt and not inline_system_prompt:
         cmd.extend(["--append-system-prompt", system_prompt])
+    stdin_text = (
+        _claude_stdin_prompt(system_prompt=system_prompt, dialogue=dialogue)
+        if inline_system_prompt
+        else dialogue
+    )
 
     # Tag the child process so any hooks it fires (e.g. claude-smart's
     # Stop hook) can detect that this is a reflexio-internal invocation
@@ -460,9 +507,11 @@ def _run_claude_stream(
     try:
         proc = subprocess.run(  # noqa: S603 — cmd is constructed from validated parts.
             cmd,
-            input=dialogue,
+            input=stdin_text,
             capture_output=True,
             text=True,
+            encoding="utf-8",
+            errors=_subprocess_text_errors(),
             timeout=timeout_seconds,
             check=False,
             env=env,
@@ -511,6 +560,8 @@ def _run_codex_stream(
             input=_codex_prompt(prompt=dialogue, system_prompt=system_prompt),
             capture_output=True,
             text=True,
+            encoding="utf-8",
+            errors=_subprocess_text_errors(),
             timeout=timeout_seconds,
             check=False,
             env=env,
@@ -550,6 +601,18 @@ def _codex_prompt(*, prompt: str, system_prompt: str) -> str:
     return f"{system_prompt}\n\n## Task\n{prompt}"
 
 
+def _claude_stdin_prompt(*, system_prompt: str, dialogue: str) -> str:
+    if not system_prompt:
+        return dialogue
+    if not dialogue:
+        return system_prompt
+    return f"{system_prompt}\n\n{dialogue}"
+
+
+def _should_inline_system_prompt(system_prompt: str) -> bool:
+    return _is_windows() and len(system_prompt) > _WINDOWS_ARGV_SYSTEM_PROMPT_LIMIT
+
+
 def _build_model_response(
     model: str,
     terminal_text: str,

reflexio/server/services/storage/sqlite_storage/_base.py

@@ -394,14 +394,19 @@ def _iso_to_epoch(iso_str: str | None) -> int:
 # stored row) with a valid value.
 _MAX_SAFE_EPOCH_TS = 253_402_300_799  # 9999-12-31T23:59:59Z
 _MIN_SAFE_EPOCH_TS = 0  # 1970-01-01T00:00:00Z
+_MIN_CONTEMPORARY_MILLISECOND_EPOCH_TS = 1_500_000_000_000
 
 
 def _epoch_to_iso(ts: int) -> str:
     """Convert a Unix timestamp (seconds) to an ISO 8601 string.
 
-    Out-of-range sentinel bounds are clamped to the representable range so that
-    callers passing "open" window bounds never trigger a ``ValueError``.
+    Plausible contemporary millisecond timestamps are normalized to seconds.
+    Lower values stay in seconds-space so documented sentinel bounds such as
+    ``to_ts=10**12`` still clamp to the representable range instead of being
+    treated as a 2001 millisecond timestamp.
     """
+    if _MIN_CONTEMPORARY_MILLISECOND_EPOCH_TS <= ts <= _MAX_SAFE_EPOCH_TS * 1000:
+        ts = ts // 1000
     clamped = max(_MIN_SAFE_EPOCH_TS, min(ts, _MAX_SAFE_EPOCH_TS))
     return datetime.fromtimestamp(clamped, tz=UTC).isoformat()
 

tests/server/llm/test_claude_code_provider.py

@@ -218,18 +218,24 @@ def test_uses_stream_json_output_format(
         assert "--verbose" in cmd
 
     def test_sets_max_retries_env(self, monkeypatch: pytest.MonkeyPatch) -> None:
-        """CLAUDE_CODE_MAX_RETRIES=3 must be passed in the subprocess env."""
+        """CLI subprocess env and stdin encoding are pinned for Windows safety."""
         mock_run = self._mock_cli(monkeypatch)
+        monkeypatch.setattr(ccp, "_is_windows", lambda: True)
         llm = ClaudeCodeLLM()
 
         llm.completion(
             model="claude-code/default",
-            messages=[{"role": "user", "content": "hi"}],
+            messages=[{"role": "user", "content": "Use an em dash — in the prompt."}],
         )
 
-        env = mock_run.call_args.kwargs["env"]
+        kwargs = mock_run.call_args.kwargs
+        env = kwargs["env"]
         assert env["CLAUDE_CODE_MAX_RETRIES"] == "3"
         assert env["CLAUDE_SMART_INTERNAL"] == "1"
+        assert kwargs["text"] is True
+        assert kwargs["encoding"] == "utf-8"
+        assert kwargs["errors"] == "replace"
+        assert kwargs["input"] == "User: Use an em dash — in the prompt."
 
     def test_system_message_goes_to_append_system_prompt_flag(
         self, monkeypatch: pytest.MonkeyPatch
@@ -251,6 +257,29 @@ def test_system_message_goes_to_append_system_prompt_flag(
         assert cmd[flag_idx + 1] == "Be terse."
         # User turn goes through stdin, not argv.
         assert mock_run.call_args.kwargs["input"] == "User: hello"
+        assert mock_run.call_args.kwargs["errors"] == "strict"
+
+    def test_large_windows_system_prompt_moves_to_stdin(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        mock_run = self._mock_cli(monkeypatch)
+        monkeypatch.setattr(ccp, "_is_windows", lambda: True)
+        llm = ClaudeCodeLLM()
+        long_system_prompt = "Use JSON only. " + ("schema-field " * 900)
+
+        llm.completion(
+            model="claude-code/default",
+            messages=[
+                {"role": "system", "content": long_system_prompt},
+                {"role": "user", "content": "Extract playbooks."},
+            ],
+        )
+
+        cmd = mock_run.call_args.args[0]
+        assert "--append-system-prompt" not in cmd
+        assert mock_run.call_args.kwargs["input"] == (
+            f"{long_system_prompt}\n\nUser: Extract playbooks."
+        )
 
     def test_no_system_message_omits_flag(
         self, monkeypatch: pytest.MonkeyPatch
@@ -335,6 +364,38 @@ def test_response_format_dict_schema_also_injected(
         flag_idx = cmd.index("--append-system-prompt")
         assert '"x"' in cmd[flag_idx + 1]
 
+    def test_large_windows_response_format_schema_moves_to_claude_stdin(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        mock_run = self._mock_cli(monkeypatch, result_text="{}")
+        monkeypatch.setattr(ccp, "_is_windows", lambda: True)
+        llm = ClaudeCodeLLM()
+        properties = {
+            f"field_{idx}": {"type": "string", "description": "required output"}
+            for idx in range(120)
+        }
+
+        llm.completion(
+            model="claude-code/default",
+            messages=[{"role": "user", "content": "Extract"}],
+            optional_params={
+                "response_format": {
+                    "type": "json_schema",
+                    "json_schema": {
+                        "schema": {"type": "object", "properties": properties}
+                    },
+                }
+            },
+        )
+
+        cmd = mock_run.call_args.args[0]
+        stdin = mock_run.call_args.kwargs["input"]
+        assert "--append-system-prompt" not in cmd
+        assert "You MUST respond with a single JSON object" in stdin
+        assert '"field_119"' in stdin
+        assert "\n\nUser: Extract" in stdin
+        assert "## Task" not in stdin
+
     def test_unsupported_params_warn_once(
         self, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
     ) -> None:
@@ -448,25 +509,81 @@ def fake_run(cmd, **kwargs):
 
         mock_run = MagicMock(side_effect=fake_run)
         monkeypatch.setenv("CLAUDE_SMART_HOST", "codex")
+        monkeypatch.setattr(ccp, "_is_windows", lambda: True)
         monkeypatch.setattr(ccp.subprocess, "run", mock_run)
         monkeypatch.setattr(ccp, "_resolve_cli_path", lambda: "/usr/local/bin/codex")
 
         response = ClaudeCodeLLM().completion(
             model="claude-code/default",
             messages=[
                 {"role": "system", "content": "Be terse."},
-                {"role": "user", "content": "ping"},
+                {"role": "user", "content": "ping — now"},
             ],
         )
 
         cmd = mock_run.call_args.args[0]
         assert cmd[:2] == ["/usr/local/bin/codex", "exec"]
         assert "-p" not in cmd
         assert "--append-system-prompt" not in cmd
-        assert mock_run.call_args.kwargs["input"] == "Be terse.\n\n## Task\nUser: ping"
-        assert mock_run.call_args.kwargs["env"]["CLAUDE_SMART_HOST"] == "codex"
+        kwargs = mock_run.call_args.kwargs
+        assert kwargs["text"] is True
+        assert kwargs["encoding"] == "utf-8"
+        assert kwargs["errors"] == "replace"
+        assert kwargs["input"] == "Be terse.\n\n## Task\nUser: ping — now"
+        assert kwargs["env"]["CLAUDE_SMART_HOST"] == "codex"
         assert response.choices[0].message.content == "codex reply"  # type: ignore[union-attr]
 
+    def test_windows_extensionless_cli_override_prefers_adjacent_cmd(
+        self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+    ) -> None:
+        bridge = tmp_path / "opencode-claude-compat"
+        bridge_cmd = tmp_path / "opencode-claude-compat.cmd"
+        bridge_cmd.write_text("@echo off\n")
+        bridge_cmd.chmod(0o755)
+
+        monkeypatch.setattr(ccp, "_is_windows", lambda: True)
+        monkeypatch.setenv(ccp.ENV_ENABLE, "1")
+        monkeypatch.setenv(ccp._ENV_CLI_PATH, str(bridge))
+
+        assert ccp._resolve_cli_path() == str(bridge_cmd)  # noqa: SLF001
+        assert is_claude_code_available()
+
+    @pytest.mark.parametrize("suffix", [".exe", ".bat"])
+    def test_windows_extensionless_cli_override_tries_common_shims(
+        self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path, suffix: str
+    ) -> None:
+        bridge = tmp_path / "claude"
+        bridge_shim = tmp_path / f"claude{suffix}"
+        bridge_shim.write_text("shim\n")
+        bridge_shim.chmod(0o755)
+
+        monkeypatch.setattr(ccp, "_is_windows", lambda: True)
+        monkeypatch.setenv(ccp.ENV_ENABLE, "1")
+        monkeypatch.setenv(ccp._ENV_CLI_PATH, str(bridge))
+
+        assert ccp._resolve_cli_path() == str(bridge_shim)  # noqa: SLF001
+
+    def test_windows_extensionless_cli_override_executes_adjacent_cmd(
+        self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+    ) -> None:
+        bridge = tmp_path / "opencode-claude-compat"
+        bridge_cmd = tmp_path / "opencode-claude-compat.cmd"
+        bridge_cmd.write_text("@echo off\n")
+        bridge_cmd.chmod(0o755)
+        mock_run = MagicMock(return_value=_fake_completed_process(_stream_json("ok")))
+
+        monkeypatch.setattr(ccp, "_is_windows", lambda: True)
+        monkeypatch.setenv(ccp._ENV_CLI_PATH, str(bridge))
+        monkeypatch.setattr(ccp.subprocess, "run", mock_run)
+
+        ClaudeCodeLLM().completion(
+            model="claude-code/default",
+            messages=[{"role": "user", "content": "hi"}],
+        )
+
+        cmd = mock_run.call_args.args[0]
+        assert cmd[0] == str(bridge_cmd)
+
     def test_codex_host_uses_compat_wrapper_with_claude_flags(
         self, monkeypatch: pytest.MonkeyPatch
     ) -> None:

tests/server/services/storage/test_sqlite_storage.py

@@ -29,7 +29,10 @@
     _true_rrf_merge,
     _vector_rank_rows,
 )
-from reflexio.server.services.storage.sqlite_storage._base import _iso_to_epoch
+from reflexio.server.services.storage.sqlite_storage._base import (
+    _epoch_to_iso,
+    _iso_to_epoch,
+)
 
 # ---------------------------------------------------------------------------
 # Helpers
@@ -63,6 +66,26 @@ def test_iso_to_epoch_treats_offsetless_timestamp_as_utc():
     assert result == expected
 
 
+def test_epoch_to_iso_preserves_second_epoch() -> None:
+    epoch_seconds = 1_700_000_000
+
+    assert _iso_to_epoch(_epoch_to_iso(epoch_seconds)) == epoch_seconds
+
+
+def test_epoch_to_iso_normalizes_plausible_millisecond_epoch() -> None:
+    epoch_ms = 1_700_000_000_123
+
+    assert _iso_to_epoch(_epoch_to_iso(epoch_ms)) == 1_700_000_000
+
+
+def test_epoch_to_iso_preserves_documented_open_bound_sentinel() -> None:
+    assert _iso_to_epoch(_epoch_to_iso(10**12)) == 253_402_300_799
+
+
+def test_epoch_to_iso_clamps_large_open_bound_sentinel() -> None:
+    assert _iso_to_epoch(_epoch_to_iso(10**18)) == 253_402_300_799
+
+
 class TestSanitizeFtsQuery:
     def test_bare_tokens_not_quoted(self):
         result = _sanitize_fts_query("user login")