# plan — workpc (compute + transport)

workpc owns: GPU, model lifecycle, per-item worker endpoints, the session manager that enforces one-warm-model-at-a-time. workpc holds NO durable state — every artifact goes to homesrv (minio) before a stage is considered done. sqlite lives on homesrv; workpc workers are stateless HTTP endpoints that read inputs by URI and write outputs to minio.

this plan assumes the homesrv plan (sqlite schema, minio buckets, orchestrator service, n8n workflow) lands in parallel. workpc can't be fully tested without homesrv's transport layer, but each worker can be developed and self-checked in isolation against a local minio + sqlite stub.

---

## phase 0 — repo hygiene

- `git init` in `/home/kami/Programs/n8n-worker/`, initial commit of current state on a `legacy/` branch. main branch starts clean.
- `.gitignore`: `.venv/`, `__pycache__/`, `dots.tts/` (submodule-ish, has its own repo), any local model paths.
- move existing workers into `legacy/` for reference, don't delete — the new code is a rewrite, not an edit.

## phase 1 — session manager (the GPU mutex)

new file: `session_manager.py`. FastAPI on `127.0.0.1:8095`.

this is the single most important new component on workpc. it owns:
- which model is currently loaded (gemma4 / siglip2 / dots.tts / comfyui-via-external)
- the llama-server subprocess lifecycle (start, health-wait, teardown)
- a GPU mutex — only one model resident at a time, enforced via a lock
- session lease tracking (who opened what, when, for timeout/reaper logic)

### endpoints

```
POST /session/open   {model: "gemma4"|"siglip2"|"dots", ttl: 3600}
  -> {session_id, port}   # port is the warm server's localhost port
  409 if another model is resident (caller retries or queues)

POST /session/close  {session_id}
  -> {ok: true}   # tears down the subprocess, releases GPU

GET  /session/active
  -> {model, session_id, opened_at, port} | null

POST /session/heartbeat  {session_id}
  -> {ok: true}   # extends the lease; stale sessions get reaped
```

### model registry

hardcoded map in the file (not config — these paths are workpc-specific):

```python
MODELS = {
    "gemma4": {
        "binary": "/mnt/D/AI/llama-cpp/llama.cpp/build/bin/llama-server",
        "model":  "/mnt/D/AI/gemma4/gemma-4-12B-it-qat-UD-Q4_K_XL.gguf",
        "mmproj": "/mnt/D/AI/gemma4/mmproj-F16.gguf",
        "port": 8090, "ctx": 4096, "args_extra": ["--jinja", "--reasoning-budget", "0"],
    },
    "siglip2": {
        "binary": "transformers",  # in-process, not subprocess
        "model":  "google/siglip2-so400m-patch16-384",
        "port": None,  # no http server, direct python call
    },
    "dots": {
        "binary": "transformers",  # in-process
        "model":  "rednote-hilab/dots.tts-base",
    },
}
```

siglip2 and dots load in-process (transformers, the rocm torch build already on workpc). gemma4 stays subprocess (llama-server, the working mtmd path). the session manager handles both kinds.

### reaper

background thread: any session without a heartbeat in `ttl` seconds gets force-closed. prevents a crashed n8n node from holding the GPU forever.

### systemd

system unit `/etc/systemd/system/manga-session.service`, `User=kami`, `ExecStart` via the workpc venv (`python -m uvicorn session_manager:app --host 127.0.0.1 --port 8095`). `--system-site-packages` venv to reuse the rocm torch build.

### self-check

`python session_manager.py`:
- open gemma4 session → active → open second → 409 → close first → open second → ok
- heartbeat: open with ttl=2, sleep 5, assert reaped via `/session/active`
- (gemma4 subprocess not actually started in self-check — mock the binary path to `/bin/sleep`)

---

## phase 2 — transport layer (minio client + sqlite status reporter)

new file: `transport.py`. shared by all workers. two concerns:

### 2a. minio artifact client

```python
def put(local_path: str, uri: str) -> str:
    """upload local file to minio at uri (e.g. 'manga/{mid}/{cid}/panels/p001.png'),
    return the full s3-style uri. raises on failure."""

def get(uri: str, local_path: str) -> str:
    """download uri to local_path, return local_path. workers pull inputs to /dev/shm
    before processing, push outputs back."""

def exists(uri: str) -> bool:
    """HEAD check. used for skip-if-exists at the worker level (defense in depth —
    the orchestrator on homesrv is the real skip authority via sqlite)."""

def put_bytes(data: bytes, uri: str) -> str:
    """for generated artifacts that don't touch disk (embeddings, json)."""
```

minio endpoint: `192.168.1.104:9000` (homesrv). creds via env, defaults match existing `storage.py`.

### 2b. sqlite status reporter

workers don't own state, but they need to report progress so n8n/the orchestrator can poll. two options:

**option A (chosen): workers don't touch sqlite directly.** they return progress in their HTTP response, the orchestrator on homesrv writes to sqlite. workers are pure compute — input URI in, output URI + metadata out. keeps the "workpc = compute only" boundary clean.

**option B (rejected): workers write to sqlite over the nginx-exposed endpoint.** adds a network-sqlite dependency to every worker, couples them to homesrv's db schema. rejected.

so `transport.py` is just minio. sqlite stays homesrv-side. workers are stateless.

### self-check

`python transport.py`: put a tempfile, get it back, assert bytes match, exists() true, missing uri exists() false. mock minio client for offline test.

---

## phase 3 — worker: OCR (stage 3)

new file: `worker_ocr.py`. FastAPI on `127.0.0.1:8001`.

### input

```json
POST /ocr
{
  "panel_uri": "s3://manga/{mid}/{cid}/panels/p001.png",
  "job_id": "...",
  "panel_id": "p001"
}
```

### process

1. `transport.get(panel_uri, /dev/shm/ocr_input.png)`
2. pytesseract.image_to_data with bbox output (`--psm 11` sparse, manga bubbles are isolated)
3. filter low-confidence (<0.5), dedupe overlapping boxes
4. cluster text lines (group words on same row within x-threshold)
5. return structured result

### output

```json
{
  "panel_id": "p001",
  "texts": [
    {"id": "t001", "content": "...", "bbox": [x,y,w,h], "confidence": 0.97}
  ]
}
```

the orchestrator (homesrv) takes this, writes to sqlite `ocr_results` table, and feeds it forward to vision + scene graph. worker doesn't persist anything — it returns the json, homesrv stores it.

### self-check

synthetic image with two text blocks, assert two texts returned with sane bboxes. no minio in self-check (local path variant of the endpoint for testing).

---

## phase 4 — worker: vision (stage 4, reworked)

new file: `worker_vision.py`. FastAPI on `127.0.0.1:8002`.

### input

```json
POST /vision
{
  "panel_uri": "...",
  "ocr_texts": [...],            // from stage 3, injected into prompt
  "known_characters": [          // from identity store, injected into prompt
    {"name": "Gojo", "description": "white hair, blindfold, black coat"}
  ],
  "panel_id": "p001",
  "session_id": "..."            // gemma4 session, opened by orchestrator via session manager
}
```

### process

1. `transport.get(panel_uri, /dev/shm/vision_input.png)`
2. build prompt: structured output request + OCR text + known characters
3. call gemma4 at `localhost:8090/v1/chat/completions` (the session manager's warm server)
4. parse structured json from response (gemma4 with `--jinja` can do json mode; enforce via prompt + json.loads with retry on parse failure)

### output (matches spec §4)

```json
{
  "panel_id": "p001",
  "characters": [
    {"local_id": "person_1", "appearance": {"hair": "...", "clothing": "...", "features": [...]}, "emotion": "...", "action": "..."}
  ],
  "scene": {"location": "...", "time": "..."}
}
```

### prompt design (the hard part)

prompt must produce parseable json. structure:

```
You are analyzing a manga panel. Known characters in this story: Gojo (white hair, blindfold, black coat).

Text found in speech bubbles: "..." , "..."

Respond in this exact JSON format:
{
  "characters": [{"local_id": "person_1", "appearance": {"hair": "", "clothing": "", "features": []}, "emotion": "", "action": ""}],
  "scene": {"location": "", "time": ""}
}
```

gemma4's reasoning output (`<|channel>thought...<channel|>`) must be stripped before json parse — reuse `_strip_thought` from legacy code.

### self-check

mock the gemma4 call (return canned json), assert parse + structured output. real gemma4 test is integration (needs session manager + model).

---

## phase 5 — worker: character identity (stage 5)

new file: `worker_identity.py`. FastAPI on `127.0.0.1:8003`.

this is the biggest new worker. it does NOT load a model itself — it uses the siglip2 session (opened via session manager) for embeddings, and queries homesrv's sqlite (via an orchestrator endpoint, not direct sqlite) for known characters.

### input

```json
POST /identity/resolve
{
  "panel_uri": "...",
  "panel_id": "p001",
  "vision_characters": [          // from stage 4
    {"local_id": "person_1", "appearance": {...}, "bbox_in_panel": [...]}
  ],
  "manga_id": "...",
  "session_id": "..."            // siglip2 session
}
```

### process

1. `transport.get(panel_uri, /dev/shm/ident_input.png)`
2. for each vision character:
   a. crop the character region from the panel (bbox from vision stage — **open question: does vision return character bboxes? currently spec §4 doesn't include them. needs adding.**)
   b. compute siglip2 image embedding of the crop
   c. query orchestrator: `GET /characters/known?manga_id=...` → list of known characters with their reference embedding URIs
   d. for each known character: `transport.get(embedding_uri, /dev/shm/ref.npy)`, cosine similarity vs the new crop embedding
   e. if best match > threshold (start 0.85, tune per-manga): assign `character_id`
   f. if below threshold: create unknown candidate → `POST /characters/create` to orchestrator, get new `character_id`, upload the crop as its first reference image

### output

```json
{
  "panel_id": "p001",
  "assignments": [
    {"local_id": "person_1", "character_id": "character_001", "confidence": 0.96}
  ],
  "new_characters": ["character_007"]
}
```

### threshold + matching strategy

- cosine similarity on siglip2 image embeddings, top-1 match
- threshold per-manga, stored in sqlite `manga_config` table (default 0.85)
- if multiple known characters are close (within 0.05 of each other), pick the best but flag low-confidence for the scene graph to handle gracefully (use "a person" in narration rather than a wrong name)

### self-check

mock siglip2 (return deterministic vectors), mock orchestrator (return known chars), assert: high-similarity → assign, low → create new. no real model in self-check.

---

## phase 6 — worker: scene graph (stage 6)

new file: `worker_scene.py`. FastAPI on `127.0.0.1:8004`.

this is a join stage, not a model stage. no GPU needed. it could run on homesrv, but keeping it on workpc for uniformity (and it's cheap enough to not need the session manager).

### input

```json
POST /scene/build
{
  "panel_id": "p001",
  "panel_uri": "...",
  "ocr_texts": [...],            // stage 3
  "vision_result": {...},        // stage 4
  "identity_assignments": [...], // stage 5
  "characters_registry": [...]   // known characters with names, for speaker attribution
}
```

### process

1. join vision characters with identity assignments → characters with real names
2. **speaker attribution** (the hard sub-problem):
   - for each OCR text block, find the nearest vision character bbox (spatial heuristic)
   - if text contains a name that ISN'T the nearest character (e.g. "Teto, come here" nearest to person_2), the speaker is likely person_2 addressing Teto
   - if text is a question/exclamation with no names, speaker = nearest character
   - if ambiguous, mark `speaker: null` (narration will quote without attribution)
   - **this is heuristic and will be wrong sometimes — accept it for v2, flag for v3 improvement with a dedicated model**
3. build scene graph

### output (matches spec §6)

```json
{
  "panel_id": "p001",
  "characters": [{"id": "character_001", "name": "Gojo Satoru"}],
  "dialogue": [{"speaker": "character_001", "text": "..."}],
  "action": "character approaches enemy"
}
```

### self-check

synthetic inputs with known spatial layout, assert speaker attribution picks nearest character. test the name-exception case.

---

## phase 7 — worker: script (stage 7, reworked)

new file: `worker_script.py`. FastAPI on `127.0.0.1:8005`.

### input

```json
POST /script
{
  "scene_graph": {...},          // stage 6
  "chapter_context": "...",      // prior scenes summary, for continuity
  "panel_id": "p001",
  "session_id": "..."            // gemma4 session (reused from vision stage if still open)
}
```

### process

1. build prompt from scene graph: characters (with names), dialogue (quoted), action, scene setting
2. call gemma4 (warm session)
3. return narration text

### output

```json
{"panel_id": "p001", "text": "..."}
```

### prompt design

```
Write narration for this manga panel. Use the characters' real names. Quote dialogue naturally. One paragraph.

Characters present: Gojo Satoru (white hair, blindfold)
Dialogue: Gojo says "..."
Action: Gojo approaches the enemy
Setting: street, night

Narration:
```

### self-check

mock gemma4, assert prompt construction includes character names + dialogue. real test is integration.

---

## phase 8 — worker: TTS (stage 8, reworked)

new file: `worker_tts.py`. FastAPI on `127.0.0.1:8006`.

reuses the dots.tts code from legacy `render_worker.py` but decoupled from layers.

### input

```json
POST /tts
{
  "text": "...",
  "speaker": "narrator",   // for now always narrator; multi-voice is v3
  "panel_id": "p001",
  "session_id": "..."      // dots session
}
```

### output

```json
{
  "audio_uri": "s3://manga/{mid}/{cid}/audio/p001.wav",
  "duration": 4.2
}
```

worker generates audio locally, `transport.put` to minio, returns the uri. no local persistence.

### self-check

mock the tts model (write a sine wave), assert upload + uri return. real test is integration.

---

## phase 9 — worker: layers (stage 9 part 1)

new file: `worker_layers.py`. FastAPI on `127.0.0.1:8007`.

wraps the comfyui layered workflow from legacy `render_worker.py`. comfyui runs as its own external process (not via session manager — it has its own server lifecycle).

### input

```json
POST /layers
{
  "panel_uri": "...",
  "panel_id": "p001",
  "num_layers": 4,
  "prompt": "..."
}
```

### output

```json
{
  "layer_uris": ["s3://manga/{mid}/{cid}/layers/p001/0.png", ...]
}
```

### self-check

mock comfyui (return canned image urls), assert upload + uri list. real test needs comfyui running.

---

## phase 10 — worker: render (stage 9 part 2, the new part)

new file: `worker_render.py`. FastAPI on `127.0.0.1:8008`.

this is the **new** rendering work — camera movement, transitions, subtitles, effects. spec §9. this is the most open-ended phase.

### input

```json
POST /render/scene
{
  "panel_uri": "...",
  "layer_uris": [...],          // from stage 9 part 1 (optional)
  "audio_uri": "...",
  "narration_text": "...",      // for subtitles
  "panel_id": "p001",
  "scene_timing": {"start": 0.0, "end": 4.2}
}
```

### process

1. pull inputs to /dev/shm
2. if layers present: parallax motion (subtle pan/zoom on separated depth layers)
3. if no layers: ken burns effect (slow zoom/pan on static panel)
4. burn subtitles from narration_text (ass format, styled)
5. transitions between scenes: crossfade (handled at the chapter assembly level, not per-scene — see below)
6. encode scene clip with ffmpeg

### output

```json
{"clip_uri": "s3://manga/{mid}/{cid}/clips/p001.mp4"}
```

### chapter assembly (separate endpoint)

```json
POST /render/assemble
{
  "clip_uris": [...],           // ordered
  "chapter_id": "..."
}
```

concat with crossfade transitions between clips, output final `chapter.mp4`, upload to minio.

### self-check

synthetic images + silent audio, assert clip produced + uploaded. real motion test is visual (manual review).

---

## phase 12 — worker: crop (stage 2, runs first)

new file: `worker_crop.py`. FastAPI on `127.0.0.1:8000`.

the panel detector. stage 2 in the pipeline (before OCR) but written last — it's the one worker the homesrv plan referenced without a contract here. no GPU (kumiko panel detection is cpu/opencv), no session manager. one page in, N panel crops out.

### input

```json
POST /crop
{
  "page_uri": "s3://manga/{mid}/{cid}/pages/p_000.png",
  "manga_id": "...",
  "chapter_id": "...",
  "page_index": 0,
  "job_id": "..."
}
```

### process

1. `transport.get(page_uri, /dev/shm/crop_input.png)`
2. kumiko panel detection (reuse the legacy kumiko path) → ordered list of panel bboxes in reading order (right-to-left, top-to-bottom for manga)
3. for each detected panel: crop from the page, `transport.put` to `manga/{mid}/{cid}/panels/pg{page_index:03d}_p{panel_index:02d}.png`
4. return the panel URIs + bboxes **in reading order** (panel_index is the within-page order; the orchestrator assigns the flat chapter-wide `panel_id`/reading order)

### output

```json
{
  "page_index": 0,
  "panels": [
    {"panel_index": 0, "uri": "s3://manga/{mid}/{cid}/panels/pg000_p00.png", "bbox": [x, y, w, h]}
  ]
}
```

`bbox` is `[x, y, w, h]` on the source page (matches the homesrv `panels.bbox` column). the orchestrator is the skip authority: it calls `/crop` once per page and skips pages that already have panel rows (defense-in-depth `transport.exists` in the worker is optional).

### self-check

synthetic 2-panel page image, assert two panels returned in reading order with sane bboxes + uploaded URIs. no real kumiko model needed if the detector is opencv-only; otherwise mock the detector.

---

## phase 11 — startup script

rewrite `start_workers.sh` to launch:
- session_manager (8095)
- worker_crop (8000)
- worker_ocr (8001)
- worker_vision (8002)
- worker_identity (8003)
- worker_scene (8004)
- worker_script (8005)
- worker_tts (8006)
- worker_layers (8007)
- worker_render (8008)

each as a background uvicorn process in a tmux session. session manager starts first, workers wait for it.

systemd units for each (system services, `User=kami`) — preferred over tmux for production, tmux for dev. provide both.

---

## open questions deferred to homesrv plan

- exact sqlite schema (characters table, ocr_results, vision_results, scene_graphs, scripts, audio, clips, jobs, stages) — defined in homesrv plan
- orchestrator endpoint contracts (`/characters/known`, `/characters/create`, stage status writes) — defined in homesrv plan
- n8n workflow shape (which nodes, how session open/close maps to stage boundaries) — defined in homesrv plan
- minio bucket layout (`manga` single bucket with key prefixes, or per-manga buckets?) — defined in homesrv plan, workpc transport.py just takes URIs

## open question flagged for your decision

- **vision stage character bboxes**: spec §4 doesn't include character bounding boxes in the vision output, but the identity worker (phase 5) needs them to crop character regions for siglip2 embeddings. add `bbox` to each character in the vision output? this is a spec amendment. my recommendation: yes, add it — gemma4 can produce bboxes if asked, and it's necessary for the identity crop. flag this in the homesrv plan too since it touches the spec.