initial commit

internal/worker/client.go

@@ -0,0 +1,186 @@
+// worker/client.go — the core-side dialer.
+//
+// Core is the CLIENT of its stt/tts worker modules: it dials them, ships
+// audio bytes (Transcribe) and reply text (Synthesize), reads back the
+// result. One Client ⇒ one conn ⇒ one mutex ⇒ no frame interleaving by
+// construction (a module needing parallelism opens N clients, but stt/tts
+// jobs are serial at single-user scale — the model is the bottleneck, not
+// the wire). ctx cancel ⇒ conn close (a half-sent frame desyncs the stream;
+// teardown is the clean recovery, a fresh Dial is the caller's job on next
+// call). Same instinct as internal/ipc/client.go.
+//
+// The daemon holds two Clients — one for the stt module, one for tts — each
+// behind a stt.Transcriber / tts.Synthesizer interface (see internal/stt,
+// internal/tts). The interface IS the swap seam: a Stub impl satisfies the
+// same interface in-process; the Remote wraps this Client. Core never sees
+// a difference.
+package worker
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"net"
+	"sync"
+	"time"
+)
+
+// Client — one connection to one worker module. NOT goroutine-safe for
+// concurrent calls on the same conn: the mutex serializes writes per
+// request so a frame doesn't interleave with another, but parallel calls
+// block on each other. Open N clients for N parallel jobs.
+type Client struct {
+	path string
+
+	mu  sync.Mutex
+	c   net.Conn
+	dial func() (net.Conn, error)
+}
+
+// Dial opens a Client to the worker socket at path. The first call lazily
+// dials; subsequent calls reuse the conn (a fresh dial happens on next call
+// after a teardown). Lazy dial keeps a worker that's restarting from
+// blocking core's startup; core attempts the dial on first use.
+func Dial(path string) *Client {
+	return &Client{
+		path: path,
+		dial: func() (net.Conn, error) {
+			return net.Dial("unix", path)
+		},
+	}
+}
+
+// Close releases the conn. Idempotent; subsequent calls re-dial on demand.
+func (c *Client) Close() error {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	if c.c == nil {
+		return nil
+	}
+	err := c.c.Close()
+	c.c = nil
+	return err
+}
+
+// Transcribe calls the transcribe verb on the connected worker module. Raises
+// ErrUnknownMethod when the worker serves synthesize only; the daemon wiring
+// has pointed this Client at the wrong socket, surfaced as a clean error.
+func (c *Client) Transcribe(ctx context.Context, req TranscribeReq) (TranscribeResp, error) {
+	var resp TranscribeResp
+	err := c.call(ctx, MethodTranscribe, req, &resp)
+	return resp, err
+}
+
+// Synthesize calls the synthesize verb.
+func (c *Client) Synthesize(ctx context.Context, req SynthesizeReq) (SynthesizeResp, error) {
+	var resp SynthesizeResp
+	err := c.call(ctx, MethodSynthesize, req, &resp)
+	return resp, err
+}
+
+func (c *Client) call(ctx context.Context, m Method, params any, out any) error {
+	body, err := marshalParams(params)
+	if err != nil {
+		return err
+	}
+	req := Request{Method: m, Params: body}
+
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	if err := c.ensureConnLocked(ctx); err != nil {
+		return err
+	}
+	// ctx cancel ⇒ close conn so a stalled peer doesn't hang the caller; a
+	// half-sent frame would desync the stream. Teardown is the clean
+	// recovery; the next call re-dials.
+	if dl, ok := ctx.Deadline(); ok {
+		_ = c.c.SetDeadline(dl)
+	} else {
+		_ = c.c.SetDeadline(time.Now().Add(defaultCallTimeout))
+	}
+	defer c.c.SetDeadline(time.Time{})
+
+	if err := writeFrame(c.c, req); err != nil {
+		c.teardownLocked()
+		return err
+	}
+	var resp Response
+	if err := readFrame(c.c, &resp); err != nil {
+		c.teardownLocked()
+		return err
+	}
+	if resp.Error != nil {
+		return hydrate(resp.Error)
+	}
+	if err := json.Unmarshal(resp.Result, out); err != nil {
+		return fmt.Errorf("worker: unmarshal result: %w", err)
+	}
+	return nil
+}
+
+func (c *Client) ensureConnLocked(ctx context.Context) error {
+	if c.c != nil {
+		return nil
+	}
+	connCh := make(chan net.Conn, 1)
+	errCh := make(chan error, 1)
+	go func() {
+		nc, err := c.dial()
+		if err != nil {
+			errCh <- err
+			return
+		}
+		connCh <- nc
+	}()
+	select {
+	case nc := <-connCh:
+		c.c = nc
+		return nil
+	case err := <-errCh:
+		return fmt.Errorf("worker: dial %s: %w", c.path, err)
+	case <-ctx.Done():
+		// the dial goroutine will finish or not; the conn if any is leaked
+		// to GC. acceptable — dial failures are rare, and a leaked closed
+		// socket is harmless.
+		return ctx.Err()
+	}
+}
+
+func (c *Client) teardownLocked() {
+	if c.c != nil {
+		_ = c.c.Close()
+		c.c = nil
+	}
+}
+
+// defaultCallTimeout — a worker that hangs > 60s is dead or stuck on a
+// model forward pass that's gone off the rails. Surface as a timeout
+// instead of hanging the daemon's tick / voice path. Production may override
+// via ctx (a long meeting-record transcription, post-MVP).
+const defaultCallTimeout = 60 * time.Second
+
+func hydrate(e *RpcError) error {
+	switch e.Code {
+	case codeUnknownMethod:
+		return fmt.Errorf("%w: %s", ErrUnknownMethod, e.Message)
+	case codeBadParams:
+		return fmt.Errorf("%w: %s", ErrBadParams, e.Message)
+	default:
+		if e.Message != "" {
+			return fmt.Errorf("worker: %s: %s", e.Code, e.Message)
+		}
+		return fmt.Errorf("worker: %s", e.Code)
+	}
+}
+
+func marshalParams(v any) (json.RawMessage, error) {
+	if v == nil {
+		return nil, nil
+	}
+	b, err := json.Marshal(v)
+	if err != nil {
+		return nil, fmt.Errorf("worker: marshal params: %w", err)
+	}
+	return b, nil
+}