fix(openclaw): address PR review follow-ups

Author: yilu331

reflexio/cli/commands/setup_cmd.py

@@ -598,6 +598,19 @@ def _remove_env_keys(env_path: Path, keys: tuple[str, ...]) -> None:
         env_path.write_text("\n".join(kept) + ("\n" if kept else ""))
 
 
+def _read_env_key(env_path: Path | None, key: str) -> str | None:
+    """Read a simple KEY=value assignment from a .env file."""
+    if env_path is None or not env_path.exists():
+        return None
+    prefix = f"{key}="
+    for line in env_path.read_text().splitlines():
+        stripped = line.strip()
+        if not stripped or stripped.startswith("#") or not stripped.startswith(prefix):
+            continue
+        return stripped[len(prefix) :].strip().strip('"').strip("'") or None
+    return None
+
+
 def _run_smart_install(plugin_dir: Path) -> None:
     """Run the plugin's first-run installer so uv/.venv land before first hook."""
     script = plugin_dir / "scripts" / "smart-install.sh"
@@ -676,7 +689,7 @@ def _install_openclaw_integration(env_path: Path) -> bool:
         # this, the gateway silently drops every plugin-side hook dispatch
         # with: '[plugins] typed hook "agent_end" blocked because non-bundled
         # plugins must set ...hooks.allowConversationAccess=true'.
-        subprocess.run(
+        access_cfg = subprocess.run(
             [
                 cli,
                 "config",
@@ -688,6 +701,12 @@ def _install_openclaw_integration(env_path: Path) -> bool:
             capture_output=True,
             text=True,
         )
+        if access_cfg.returncode != 0:
+            typer.echo(
+                "Error: could not persist openClaw conversation-access permission: "
+                f"{access_cfg.stderr or access_cfg.stdout}"
+            )
+            raise typer.Exit(1)
     except subprocess.CalledProcessError as exc:
         typer.echo(f"Error: openclaw command failed: {exc.stderr or exc.stdout}")
         raise typer.Exit(1) from exc
@@ -733,7 +752,7 @@ def _uninstall_openclaw(env_path: Path | None = None, purge: bool = False) -> No
         "This will remove the Reflexio integration from openClaw. Continue?",
         abort=True,
     )
-    cli = shutil.which("openclaw")
+    cli = _read_env_key(env_path, "OPENCLAW_BIN") or shutil.which("openclaw")
     if cli:
         subprocess.run(
             [cli, "plugins", "disable", _OPENCLAW_PLUGIN_ID],
@@ -748,7 +767,10 @@ def _uninstall_openclaw(env_path: Path | None = None, purge: bool = False) -> No
             text=True,
         )
     else:
-        typer.echo("Warning: openclaw CLI not found on PATH, skipping plugin removal")
+        typer.echo(
+            "Warning: openclaw CLI not found in OPENCLAW_BIN or PATH, "
+            "skipping plugin removal"
+        )
 
     if env_path is not None:
         _remove_env_keys(env_path, ("OPENCLAW_BIN", "OPENCLAW_SMART_USE_LOCAL_CLI"))
@@ -832,6 +854,13 @@ def openclaw(
         typer.echo("Error: could not locate or create a .env file")
         raise typer.Exit(1)
 
+    if repair and (uninstall or purge):
+        typer.echo("Error: --repair cannot be combined with --uninstall or --purge")
+        raise typer.Exit(1)
+    if purge and not uninstall:
+        typer.echo("Error: --purge requires --uninstall")
+        raise typer.Exit(1)
+
     if repair:
         _repair_openclaw()
         return

reflexio/integrations/openclaw/plugin/scripts/backend-service.sh

@@ -49,11 +49,11 @@ PLUGIN_ROOT="$(cd "$HERE/.." && pwd)"
 
 # Pin the openclaw CLI explicitly so the reflexio backend's openclaw_provider
 # can find it from a hook context whose PATH lacks the user's normal CLI dir.
-if [ -z "${OPENCLAW_SMART_CLI_PATH:-}" ]; then
+if [ -z "${OPENCLAW_BIN:-}" ]; then
   if _oc_cli_path=$(command -v openclaw 2>/dev/null) && [ -n "$_oc_cli_path" ]; then
-    export OPENCLAW_SMART_CLI_PATH="$_oc_cli_path"
+    export OPENCLAW_BIN="$_oc_cli_path"
   elif [ -x "$HOME/.local/bin/openclaw" ]; then
-    export OPENCLAW_SMART_CLI_PATH="$HOME/.local/bin/openclaw"
+    export OPENCLAW_BIN="$HOME/.local/bin/openclaw"
   fi
   unset _oc_cli_path
 fi

reflexio/integrations/openclaw/plugin/src/openclaw_smart/cli.py

@@ -523,24 +523,26 @@ def cmd_clear_all(args: argparse.Namespace) -> int:
             )
             return stop_rc or 1
 
+    clear_all_failed = False
     removed_targets = 0
     try:
         for target in targets:
             if _remove_clear_all_target(target):
                 removed_targets += 1
     except (OSError, _ClearAllError) as exc:
         sys.stderr.write(f"error: could not remove reflexio data: {exc}\n")
-        return 1
+        clear_all_failed = True
 
     removed_buffers = 0
     root = state.state_dir()
-    if root.is_dir():
+    if not clear_all_failed and root.is_dir():
         for buf in root.glob("*.jsonl"):
             try:
                 buf.unlink()
                 removed_buffers += 1
             except OSError as exc:
                 sys.stderr.write(f"warning: could not remove {buf}: {exc}\n")
+                clear_all_failed = True
 
     start_rc = 0
     if was_running:
@@ -556,7 +558,7 @@ def cmd_clear_all(args: argparse.Namespace) -> int:
         f"Cleared reflexio: {target_summary}. "
         f"Removed {removed_buffers} local session buffer(s).\n"
     )
-    return start_rc or 0
+    return start_rc or (1 if clear_all_failed else 0)
 
 
 def _build_parser() -> argparse.ArgumentParser:

reflexio/integrations/openclaw/plugin/src/openclaw_smart/publish.py

@@ -8,14 +8,37 @@
 
 from __future__ import annotations
 
+import contextlib
+from collections.abc import Iterator
 from typing import Literal
 
 from openclaw_smart import state
 from openclaw_smart.reflexio_adapter import Adapter
 
+try:
+    import fcntl  # POSIX only — Windows falls back to no publish lock.
+except ImportError:  # pragma: no cover — non-POSIX platforms
+    fcntl = None  # type: ignore[assignment]
+
 PublishStatus = Literal["nothing", "ok", "failed"]
 
 
+@contextlib.contextmanager
+def _session_publish_lock(session_id: str) -> Iterator[None]:
+    """Serialize read-publish-watermark for one session buffer."""
+    lock_path = state.publish_lock_path(session_id)
+    if lock_path is None or fcntl is None:
+        yield
+        return
+    lock_path.parent.mkdir(parents=True, exist_ok=True)
+    with lock_path.open("a", encoding="utf-8") as fh:
+        fcntl.flock(fh.fileno(), fcntl.LOCK_EX)
+        try:
+            yield
+        finally:
+            fcntl.flock(fh.fileno(), fcntl.LOCK_UN)
+
+
 def publish_unpublished(
     *,
     session_id: str,
@@ -54,19 +77,20 @@ def publish_unpublished(
             was unreachable. On ``"failed"`` the watermark is not advanced,
             so the next hook retries the same batch.
     """
-    records = state.read_all(session_id)
-    _, interactions = state.unpublished_slice(records)
-    if not interactions:
-        return ("nothing", 0)
-    client = adapter if adapter is not None else Adapter()
-    ok = client.publish(
-        session_id=session_id,
-        project_id=project_id,
-        interactions=interactions,
-        force_extraction=force_extraction,
-        skip_aggregation=skip_aggregation,
-    )
-    if ok:
-        state.append(session_id, {"published_up_to": len(records)})
-        return ("ok", len(interactions))
-    return ("failed", len(interactions))
+    with _session_publish_lock(session_id):
+        records = state.read_all(session_id)
+        _, interactions = state.unpublished_slice(records)
+        if not interactions:
+            return ("nothing", 0)
+        client = adapter if adapter is not None else Adapter()
+        ok = client.publish(
+            session_id=session_id,
+            project_id=project_id,
+            interactions=interactions,
+            force_extraction=force_extraction,
+            skip_aggregation=skip_aggregation,
+        )
+        if ok:
+            state.append(session_id, {"published_up_to": len(records)})
+            return ("ok", len(interactions))
+        return ("failed", len(interactions))

reflexio/integrations/openclaw/plugin/src/openclaw_smart/state.py

@@ -20,8 +20,9 @@
 import logging
 import os
 import re
+from collections.abc import Iterable
 from pathlib import Path
-from typing import Any, Iterable
+from typing import Any
 
 try:
     import fcntl  # POSIX only — Windows hooks fall back to append-without-lock.
@@ -115,6 +116,14 @@ def injected_path(session_id: str) -> Path | None:
     return state_dir() / f"{sid}.injected.jsonl"
 
 
+def publish_lock_path(session_id: str) -> Path | None:
+    """Return the per-session publish lock path, or ``None`` if unsafe."""
+    sid = _safe_session_id(session_id)
+    if sid is None:
+        return None
+    return state_dir() / f"{sid}.publish.lock"
+
+
 def append_injected(session_id: str, entries: Iterable[dict[str, Any]]) -> None:
     """Append citation-registry entries to the per-session injected-items file.
 
@@ -264,15 +273,18 @@ def unpublished_slice(
     turns: list[dict[str, Any]] = []
     for idx, rec in enumerate(records):
         if "published_up_to" in rec:
-            published = rec["published_up_to"]
+            marker = rec.get("published_up_to")
+            if isinstance(marker, int) and marker >= 0:
+                published = marker
             pending_tools = []
             turns = []
             continue
         if idx < published:
             continue
         role = rec.get("role")
         if role == "Assistant_tool":
-            tool_input = rec.get("tool_input") or {}
+            raw_tool_input = rec.get("tool_input")
+            tool_input = raw_tool_input if isinstance(raw_tool_input, dict) else {}
             tool_output = rec.get("tool_output") or ""
             tool_entry: dict[str, Any] = {
                 "tool_name": rec.get("tool_name", ""),

reflexio/integrations/openclaw/plugin/tests/integration/test_publish_to_local_reflexio_integration.py

@@ -1,7 +1,7 @@
 """Integration: publish buffered turns to a local SQLite-backed reflexio.
 
 Requires a reflexio backend reachable at ``REFLEXIO_URL`` (default
-``http://localhost:8071/``). Skipped automatically when the backend is
+``http://localhost:8081/``). Skipped automatically when the backend is
 unreachable so the suite is portable across machines.
 """
 
@@ -16,7 +16,7 @@
 
 
 def _reflexio_url() -> str:
-    return os.environ.get("REFLEXIO_URL", "http://localhost:8071/")
+    return os.environ.get("REFLEXIO_URL", "http://localhost:8081/")
 
 
 def _backend_alive(url: str) -> bool:

reflexio/integrations/openclaw/plugin/tests/integration/test_search_inject_integration.py

@@ -24,7 +24,7 @@
 
 
 def _reflexio_url() -> str:
-    return os.environ.get("REFLEXIO_URL", "http://localhost:8071/")
+    return os.environ.get("REFLEXIO_URL", "http://localhost:8081/")
 
 
 def _backend_alive(url: str) -> bool:

reflexio/integrations/openclaw/plugin/tests/test_cli.py

@@ -3,10 +3,9 @@
 from __future__ import annotations
 
 from argparse import Namespace
-from unittest.mock import MagicMock, patch
+from unittest.mock import patch
 
 import pytest
-
 from openclaw_smart import cli
 
 
@@ -127,6 +126,32 @@ def test_cmd_clear_all_with_yes_proceeds(monkeypatch, tmp_path):
     assert not (sessions / "old.jsonl").exists()
 
 
+def test_cmd_clear_all_restarts_backend_after_delete_failure(tmp_path):
+    target = cli._ClearAllTarget(
+        path=tmp_path / "reflexio-openclaw-test",
+        kind="dir",
+        label="test target",
+    )
+    service_calls: list[str] = []
+
+    def fake_run_service(_script, command) -> int:  # noqa: ANN001
+        service_calls.append(command)
+        return 0
+
+    with patch(
+        "openclaw_smart.cli._resolve_clear_all_targets", return_value=[target]
+    ), patch("openclaw_smart.cli._service_status", return_value="running on 8071"), patch(
+        "openclaw_smart.cli._remove_clear_all_target",
+        side_effect=cli._ClearAllError("boom"),
+    ), patch(
+        "openclaw_smart.cli._run_service", side_effect=fake_run_service
+    ):
+        rc = cli.cmd_clear_all(Namespace(yes=True))
+
+    assert rc == 1
+    assert service_calls == ["stop", "start"]
+
+
 def test_build_parser_accepts_show():
     parser = cli._build_parser()
     args = parser.parse_args(["show"])

reflexio/integrations/openclaw/plugin/tests/test_events_session_end.py

@@ -33,5 +33,9 @@ def test_handle_falls_back_to_session_id_key():
         return_value="proj-y",
     ):
         session_end.handle({"sessionId": "s2"})
+        pub.publish_unpublished.assert_called_once()
         kwargs = pub.publish_unpublished.call_args[1]
         assert kwargs["session_id"] == "s2"
+        assert kwargs["project_id"] == "proj-y"
+        assert kwargs["force_extraction"] is True
+        assert kwargs["skip_aggregation"] is False

reflexio/integrations/openclaw/plugin/tests/test_publish.py

@@ -0,0 +1,49 @@
+"""Tests for openclaw_smart.publish."""
+
+from __future__ import annotations
+
+import json
+
+import pytest
+from openclaw_smart import publish, state
+
+
+@pytest.fixture(autouse=True)
+def isolate_state_dir(monkeypatch, tmp_path):
+    sessions = tmp_path / "sessions"
+    monkeypatch.setenv("OPENCLAW_SMART_STATE_DIR", str(sessions))
+    return sessions
+
+
+class _Adapter:
+    def __init__(self) -> None:
+        self.calls = 0
+
+    def publish(self, **_kwargs) -> bool:  # noqa: ANN003
+        self.calls += 1
+        return True
+
+
+def test_publish_unpublished_serializes_with_lock_and_stamps_watermark(
+    isolate_state_dir,
+):
+    state.append("s1", {"role": "User", "content": "hi"})
+    adapter = _Adapter()
+
+    status, count = publish.publish_unpublished(
+        session_id="s1",
+        project_id="proj",
+        force_extraction=False,
+        skip_aggregation=False,
+        adapter=adapter,
+    )
+
+    assert (status, count) == ("ok", 1)
+    assert adapter.calls == 1
+    assert (isolate_state_dir / "s1.publish.lock").exists()
+    records = [
+        json.loads(line)
+        for line in (isolate_state_dir / "s1.jsonl").read_text().splitlines()
+    ]
+    assert records[-1] == {"published_up_to": 1}
+