Skip to content

zettelkasten.synapse.worker_client

zettelkasten.synapse.worker_client

Client for the long-lived WARM synapse READ worker.

Manages a single persistent python -m zettelkasten.synapse.worker subprocess and serializes queries to it over the worker's line protocol. Keeping the worker warm across queries is the repeat-query win (see :mod:zettelkasten.synapse.worker): only the first query pays the cold model load + index build; the rest reuse the warm state.

Crash isolation is preserved: a native crash kills only the worker process, which this client detects (the stdout pipe reaches EOF and poll() reports a signal) and transparently respawns on the next query. Callers distinguish outcomes by exception type:

  • :class:WorkerCrash — the worker died on a signal (native fault). The query that triggered it should NOT be retried in-process; surface a clean error.
  • :class:WorkerTimeout — the query exceeded its budget (cold rebuild / hang); the worker was killed.
  • :class:WorkerError — spawn/protocol failure. The caller should FALL BACK to the one-shot subprocess path (the proven, always-cold route).

Queries are serialized through a single lock: the worker is single-threaded and kglite is not thread-safe, so concurrent synapse reads queue rather than race. Cross-store reads are heavy and infrequent, so this is the correct trade.

WorkerError

Bases: RuntimeError

A recoverable worker failure — the caller should fall back to one-shot.

Source code in zettelkasten/synapse/worker_client.py 
48
49
class WorkerError(RuntimeError):
    """A recoverable worker failure — the caller should fall back to one-shot."""

WorkerCrash

Bases: WorkerError

The worker died on a signal (almost certainly a native kglite/model2vec fault).

Source code in zettelkasten/synapse/worker_client.py 
52
53
54
55
56
57
58
class WorkerCrash(WorkerError):
    """The worker died on a signal (almost certainly a native kglite/model2vec fault)."""

    def __init__(self, signal: int, trace: str = "") -> None:
        super().__init__(f"synapse worker crashed with signal {signal} (native fault)")
        self.signal = signal
        self.trace = trace

WorkerTimeout

Bases: WorkerError

A query exceeded its wall-clock budget; the worker was killed.

Source code in zettelkasten/synapse/worker_client.py 
61
62
63
64
65
66
class WorkerTimeout(WorkerError):
    """A query exceeded its wall-clock budget; the worker was killed."""

    def __init__(self, timeout: float) -> None:
        super().__init__(f"synapse worker query exceeded {timeout:.0f}s")
        self.timeout = timeout

SynapseWorker

Owns one persistent worker subprocess and serializes queries to it.

Source code in zettelkasten/synapse/worker_client.py 
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
class SynapseWorker:
    """Owns one persistent worker subprocess and serializes queries to it."""

    def __init__(self, cmd: "list[str] | None" = None, *, spawn_timeout: float = 60.0) -> None:
        # ``cmd`` is overridable so tests can point at a stub script; production
        # uses the real worker module.
        self._cmd = cmd or [sys.executable, "-m", "zettelkasten.synapse.worker"]
        self._spawn_timeout = spawn_timeout
        self._lock = threading.Lock()
        self._proc: "subprocess.Popen[str] | None" = None
        self._stdout_q: "queue.Queue[object]" = queue.Queue()
        self._stderr_buf: "deque[str]" = deque(maxlen=100)

    # ── lifecycle ────────────────────────────────────────────────────────────

    def _alive(self) -> bool:
        return self._proc is not None and self._proc.poll() is None

    def _spawn(self) -> None:
        workspace = os.environ.get("ANGELO_WORKSPACE") or os.environ.get("CLAUDE_PROJECT_DIR") or os.getcwd()
        try:
            proc = subprocess.Popen(
                self._cmd,
                stdin=subprocess.PIPE,
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
                text=True,
                bufsize=1,  # line-buffered
                cwd=workspace,
                env=os.environ.copy(),
            )
        except Exception as exc:  # noqa: BLE001 — spawn failure => fall back to one-shot
            raise WorkerError(f"could not launch synapse worker: {exc}") from exc

        self._proc = proc
        self._stdout_q = queue.Queue()
        self._stderr_buf = deque(maxlen=100)
        threading.Thread(
            target=self._drain_stdout, args=(proc.stdout, self._stdout_q), daemon=True
        ).start()
        threading.Thread(
            target=self._drain_stderr, args=(proc.stderr, self._stderr_buf), daemon=True
        ).start()

        # Wait for the readiness line so a worker that fails to import surfaces as
        # a clean spawn error (=> fall back) rather than hanging the first query.
        deadline = time.monotonic() + self._spawn_timeout
        while True:
            remaining = deadline - time.monotonic()
            if remaining <= 0:
                self._kill()
                raise WorkerError("synapse worker did not become ready in time")
            try:
                line = self._stdout_q.get(timeout=remaining)
            except queue.Empty:
                self._kill()
                raise WorkerError("synapse worker did not become ready in time")
            if line is _EOF:
                trace = "\n".join(self._stderr_buf)
                self._kill()
                raise WorkerError(f"synapse worker exited before ready:\n{trace}")
            if isinstance(line, str) and line.strip() == READY_LINE:
                return
            # Ignore any other startup chatter on stdout.

    @staticmethod
    def _drain_stdout(pipe: Any, q: "queue.Queue[object]") -> None:
        try:
            for line in pipe:
                q.put(line.rstrip("\n"))
        except Exception:  # noqa: BLE001 — pipe torn down on kill; nothing to do
            pass
        finally:
            q.put(_EOF)

    @staticmethod
    def _drain_stderr(pipe: Any, buf: "deque[str]") -> None:
        try:
            for line in pipe:
                buf.append(line.rstrip("\n"))
        except Exception:  # noqa: BLE001
            pass

    def _kill(self) -> None:
        proc = self._proc
        self._proc = None
        if proc is None:
            return
        try:
            if proc.poll() is None:
                proc.kill()
            proc.wait(timeout=5)
        except Exception:  # noqa: BLE001 — best-effort teardown
            pass
        for stream in (proc.stdin, proc.stdout, proc.stderr):
            try:
                if stream is not None:
                    stream.close()
            except Exception:  # noqa: BLE001
                pass

    def shutdown(self) -> None:
        """Terminate the worker (e.g. on server exit). Idempotent."""
        with self._lock:
            self._kill()

    # ── query ────────────────────────────────────────────────────────────────

    def query(self, payload: dict, timeout: float) -> dict:
        """Run one search/frame ``payload`` on the warm worker; return its result dict.

        Spawns the worker on first use (or after a crash). Raises
        :class:`WorkerCrash` / :class:`WorkerTimeout` / :class:`WorkerError` on the
        respective failure modes (see the module docstring); the worker is killed
        before raising so the next call starts fresh.
        """
        with self._lock:
            if not self._alive():
                self._spawn()
            assert self._proc is not None and self._proc.stdin is not None

            try:
                self._proc.stdin.write(json.dumps(payload) + "\n")
                self._proc.stdin.flush()
            except (BrokenPipeError, OSError) as exc:
                self._kill()
                raise WorkerError(f"synapse worker stdin closed: {exc}") from exc

            deadline = time.monotonic() + timeout
            while True:
                remaining = deadline - time.monotonic()
                if remaining <= 0:
                    self._kill()
                    raise WorkerTimeout(timeout)
                try:
                    line = self._stdout_q.get(timeout=remaining)
                except queue.Empty:
                    self._kill()
                    raise WorkerTimeout(timeout)
                if line is _EOF:
                    rc = self._proc.poll() if self._proc else None
                    trace = "\n".join(self._stderr_buf)
                    self._kill()
                    if rc is not None and rc < 0:
                        raise WorkerCrash(-rc, trace)
                    raise WorkerError(f"synapse worker exited (rc={rc}):\n{trace}")
                if isinstance(line, str) and line.startswith(RESULT_PREFIX):
                    try:
                        return json.loads(line[len(RESULT_PREFIX):])
                    except json.JSONDecodeError as exc:
                        self._kill()
                        raise WorkerError(f"synapse worker sent malformed result: {exc}") from exc

shutdown 

shutdown() -> None

Terminate the worker (e.g. on server exit). Idempotent.

Source code in zettelkasten/synapse/worker_client.py 
170
171
172
173
def shutdown(self) -> None:
    """Terminate the worker (e.g. on server exit). Idempotent."""
    with self._lock:
        self._kill()

query 

query(payload: dict, timeout: float) -> dict

Run one search/frame payload on the warm worker; return its result dict.

Spawns the worker on first use (or after a crash). Raises :class:WorkerCrash / :class:WorkerTimeout / :class:WorkerError on the respective failure modes (see the module docstring); the worker is killed before raising so the next call starts fresh.

Source code in zettelkasten/synapse/worker_client.py 
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
def query(self, payload: dict, timeout: float) -> dict:
    """Run one search/frame ``payload`` on the warm worker; return its result dict.

    Spawns the worker on first use (or after a crash). Raises
    :class:`WorkerCrash` / :class:`WorkerTimeout` / :class:`WorkerError` on the
    respective failure modes (see the module docstring); the worker is killed
    before raising so the next call starts fresh.
    """
    with self._lock:
        if not self._alive():
            self._spawn()
        assert self._proc is not None and self._proc.stdin is not None

        try:
            self._proc.stdin.write(json.dumps(payload) + "\n")
            self._proc.stdin.flush()
        except (BrokenPipeError, OSError) as exc:
            self._kill()
            raise WorkerError(f"synapse worker stdin closed: {exc}") from exc

        deadline = time.monotonic() + timeout
        while True:
            remaining = deadline - time.monotonic()
            if remaining <= 0:
                self._kill()
                raise WorkerTimeout(timeout)
            try:
                line = self._stdout_q.get(timeout=remaining)
            except queue.Empty:
                self._kill()
                raise WorkerTimeout(timeout)
            if line is _EOF:
                rc = self._proc.poll() if self._proc else None
                trace = "\n".join(self._stderr_buf)
                self._kill()
                if rc is not None and rc < 0:
                    raise WorkerCrash(-rc, trace)
                raise WorkerError(f"synapse worker exited (rc={rc}):\n{trace}")
            if isinstance(line, str) and line.startswith(RESULT_PREFIX):
                try:
                    return json.loads(line[len(RESULT_PREFIX):])
                except json.JSONDecodeError as exc:
                    self._kill()
                    raise WorkerError(f"synapse worker sent malformed result: {exc}") from exc