Hynes/Void-Homelab
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
f8beff870216f3ee3a9bc1353ea67a2ae354857f
Void-Homelab/docs/superpowers/plans/2026-06-01-void-v2-plan4-workers.md
root c4663992ec docs: Plan 4 implementation plan 
Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
2026-06-01 04:39:55 +10:00

2122 lines
72 KiB
Markdown
Raw Blame History

# Void 2.0 — Plan 4: Python void-workers (heavy ML ingest)
> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
**Goal:** Stand up the Python `void-workers` service alongside the existing Node `void-server` on CT 311. Workers claim jobs from the same pg-boss queue via native SQL (`SELECT ... FOR UPDATE SKIP LOCKED`) and process four kinds: `extract.pdf`, `extract.image`, `ingest.video`, `sync.source_doc`. Whisper detects CUDA at startup and falls back to CPU when the HA replica is running on Z3 (no GPU).
**Architecture:** Single CT, two systemd units (`void-server.service` + `void-workers.service`). Shared blob store at `/var/lib/void/blobs`. Shared Postgres at CT 310. Python venv at `/opt/void-workers/venv`. Per-kind handlers run as threads inside one Python process; concurrency configured via env. Node-side hooks: capture.js routes YouTube URLs to `ingest.video`; blob worker enqueues `extract.pdf`/`extract.image` based on `kind`.
**Tech Stack:** Python 3.12, psycopg 3, pdftotext + Tesseract + Poppler + Pillow, faster-whisper (CTranslate2 backend), yt-dlp, ffmpeg, structlog, pytest, vitest (existing), node-cron (new Node dep for Phase D).
**Spec:** `docs/superpowers/specs/2026-06-01-void-v2-plan4-workers.md`.
---
## Out of scope (Plan 5+)
- MCP server surface, Companion chat (Plan 5).
- Sacred Valley widgets (Plan 6).
- Speaker diarization, per-language Tesseract data, larger Whisper models with proper GPU budgeting.
- "Re-transcribe" UI action.
## Conventions
1. **TDD.** Write the failing test first. Run it red. Implement. Run it green. Commit.
2. **Worker handler shape:** `def handler(job_data: dict) -> dict` — return the success output, raise for retry.
3. **Worker name strings:** `extract.pdf`, `extract.image`, `ingest.video`, `sync.source_doc`. Exact strings — used in Node enqueue calls AND Python `SUBSCRIBE` registrations.
4. **All Python repo writes pass `actor={'kind': 'worker', 'id': None}`** when emitting audit_log rows.
5. **Commit per task.**
6. **Snapshot CT 310 + CT 311 before any phase that touches infra** (the standing rule). Phase A is code-only. Phase B installs system deps. Phase C resizes the CT + adds GPU passthrough — explicit snapshot before that one.
---
## File structure (what this plan creates)
```
workers/ # NEW — Python service lives in same repo
pyproject.toml # Python 3.12 + deps (psycopg, structlog, ...)
README.md # short ops notes
void_workers/
__init__.py
config.py # env loader
log.py # structlog setup
boss.py # claim / finish / fail / retry against pgboss tables
repo.py # psycopg shim for refs / source_docs updates + audit
safe_fetch.py # Python port of lib/ingest/safe_fetch.js
runner.py # main: load handlers, spawn loops, signal handlers
handlers/
__init__.py
echo.py # trivial — proves harness end-to-end
pdf.py # extract.pdf (Phase B)
image.py # extract.image (Phase B)
video.py # ingest.video (Phase C)
sourcedoc.py # sync.source_doc (Phase D)
model.py # Whisper loader (CUDA detect + CPU fallback) — Phase C
tests/
conftest.py # shared fixtures (DB url, pgboss schema reset, sample blobs)
fixtures/
born_digital.pdf # tiny PDF with embedded text
scanned.pdf # tiny PDF that is a single page of an image
eng_text.png # PNG with known text baked in
test_boss.py
test_echo.py
test_pdf.py
test_image.py
test_video.py
test_sourcedoc.py
test_safe_fetch.py
deploy/
void-workers.service # NEW systemd unit
push-workers.sh # NEW — sibling to push.sh
README.md # extend: workers bootstrap section
lib/ # Node-side changes (already exists)
jobs/workers/blob.js # MODIFY: after creating ref, enqueue extract.pdf/extract.image
api/routes/capture.js # MODIFY: detect YouTube/Vimeo, enqueue ingest.video
cron/sync_source_docs.js # NEW — node-cron, fires sync.source_doc daily
package.json # ADD node-cron, bump version in Phase D
server.js # bump VERSION + boot the cron in Phase D
CHANGELOG.md # Phase D entry
tests/server.test.js # version assertion bump
docs/plan-4-complete.md # Phase D
```
---
## Phase A — Python harness + systemd unit
### Task A1: Workers skeleton + venv install on the dev box
**Files:**
- Create: `workers/pyproject.toml`
- Create: `workers/void_workers/__init__.py`
- Create: `workers/void_workers/config.py`
- Create: `workers/void_workers/log.py`
- Create: `workers/README.md`
- [ ] **Step 1:** Create `workers/pyproject.toml`:
```toml
[project]
name = "void-workers"
version = "0.1.0"
requires-python = ">=3.12"
dependencies = [
"psycopg[binary,pool]>=3.2",
"structlog>=24.1",
]
[project.optional-dependencies]
pdf = ["pdfplumber>=0.11", "pytesseract>=0.3.13", "pillow>=10.3"]
image = ["pytesseract>=0.3.13", "pillow>=10.3"]
video = ["yt-dlp>=2024.10.0", "faster-whisper>=1.0.3"]
test = ["pytest>=8.0", "pytest-asyncio>=0.23"]
all = ["void-workers[pdf,image,video,test]"]
[tool.setuptools.packages.find]
where = ["."]
include = ["void_workers*"]
```
- [ ] **Step 2:** Create `workers/void_workers/__init__.py`:
```python
__version__ = "0.1.0"
```
- [ ] **Step 3:** Create `workers/void_workers/config.py`:
```python
import os
def env(name, default=None, required=False):
v = os.environ.get(name, default)
if required and v is None:
raise RuntimeError(f"env {name} is required")
return v
def env_int(name, default):
return int(os.environ.get(name, default))
DATABASE_URL = env("DATABASE_URL", required=True)
BLOB_ROOT = env("BLOB_ROOT", "/var/lib/void/blobs")
WHISPER_MODEL = env("WHISPER_MODEL", "small.en")
WHISPER_CACHE = env("WHISPER_CACHE", "/var/lib/void/whisper-models")
ALLOW_PRIVATE = env("VOID_INGEST_ALLOW_PRIVATE", "false") == "true"
CONCURRENCY = {
"extract.pdf": env_int("VOID_CONCURRENCY_EXTRACT_PDF", 2),
"extract.image": env_int("VOID_CONCURRENCY_EXTRACT_IMAGE", 2),
"ingest.video": env_int("VOID_CONCURRENCY_INGEST_VIDEO", 1),
"sync.source_doc": env_int("VOID_CONCURRENCY_SYNC_SOURCE_DOC", 1),
"echo": env_int("VOID_CONCURRENCY_ECHO", 1),
}
POLL_INTERVAL_MS = env_int("VOID_POLL_INTERVAL_MS", 1000)
```
- [ ] **Step 4:** Create `workers/void_workers/log.py`:
```python
import logging
import structlog
def init():
structlog.configure(
processors=[
structlog.contextvars.merge_contextvars,
structlog.processors.add_log_level,
structlog.processors.TimeStamper(fmt="iso"),
structlog.processors.JSONRenderer(),
],
wrapper_class=structlog.make_filtering_bound_logger(logging.INFO),
)
return structlog.get_logger()
log = init()
```
- [ ] **Step 5:** Create `workers/README.md`:
```markdown
# void-workers
Python ML ingest service alongside void-server (Node). Sibling of `lib/`.
## Local dev
```bash
cd workers
python3.12 -m venv .venv
. .venv/bin/activate
pip install -e ".[all]"
export DATABASE_URL="postgres://..."
python -m void_workers.runner
```
## Tests
```bash
pip install -e ".[test,all]"
pytest -v
```
See `../docs/superpowers/plans/2026-06-01-void-v2-plan4-workers.md` for the full plan.
```
- [ ] **Step 6:** Create the venv + install:
```bash
cd /project/src/void-v2/workers
python3.12 -m venv .venv
. .venv/bin/activate
pip install -e ".[all]"
python -c "from void_workers import __version__; print(__version__)"
```
Expected: `0.1.0`.
- [ ] **Step 7: Commit.**
```bash
cd /project/src/void-v2
git add workers/pyproject.toml workers/README.md workers/void_workers/__init__.py workers/void_workers/config.py workers/void_workers/log.py
git commit -m "feat(workers): Python skeleton + config + structlog"
```
### Task A2: pgboss claim/finish helpers in `boss.py`
**Files:**
- Create: `workers/void_workers/boss.py`
- Create: `workers/tests/conftest.py`
- Create: `workers/tests/test_boss.py`
- [ ] **Step 1:** Create `workers/tests/conftest.py`:
```python
import os
import pytest
import psycopg
DB_URL = os.environ["DATABASE_URL"]
@pytest.fixture
def conn():
with psycopg.connect(DB_URL, autocommit=True) as c:
yield c
@pytest.fixture(autouse=True)
def reset_pgboss(conn):
"""Drop and recreate the pgboss schema before each test."""
conn.execute("DROP SCHEMA IF EXISTS pgboss CASCADE")
# We do NOT recreate it — pg-boss's first start() makes the schema.
# Tests that need it call ensure_pgboss(conn).
yield
def ensure_pgboss(conn):
"""Bring pgboss schema up by shelling out to Node's queue.start()."""
# In tests we cannot rely on Node — so we create the bare minimum
# tables we touch. Real pgboss tables are richer; we only need the
# columns boss.py reads and writes.
conn.execute("CREATE SCHEMA IF NOT EXISTS pgboss")
conn.execute("""
CREATE TABLE IF NOT EXISTS pgboss.job (
id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
name text NOT NULL,
priority int NOT NULL DEFAULT 0,
data jsonb,
state text NOT NULL DEFAULT 'created',
retry_limit int NOT NULL DEFAULT 5,
retry_count int NOT NULL DEFAULT 0,
retry_delay int NOT NULL DEFAULT 10,
retry_backoff boolean NOT NULL DEFAULT true,
start_after timestamptz NOT NULL DEFAULT now(),
started_on timestamptz,
completed_on timestamptz,
created_on timestamptz NOT NULL DEFAULT now(),
output jsonb
)
""")
@pytest.fixture
def boss_ready(conn):
ensure_pgboss(conn)
yield conn
```
- [ ] **Step 2:** Write the failing test `workers/tests/test_boss.py`:
```python
import json
from void_workers.boss import Boss
def test_claim_returns_none_when_no_jobs(boss_ready):
boss = Boss(dsn=__import__("void_workers.config", fromlist=["DATABASE_URL"]).DATABASE_URL)
assert boss.claim("echo") is None
def test_claim_then_complete_round_trip(boss_ready):
boss_ready.execute(
"INSERT INTO pgboss.job(name, data) VALUES('echo', %s)",
(json.dumps({"ping": 1}),)
)
boss = Boss(dsn=__import__("void_workers.config", fromlist=["DATABASE_URL"]).DATABASE_URL)
job = boss.claim("echo")
assert job is not None
assert job["data"] == {"ping": 1}
boss.complete(job["id"], {"pong": 1})
row = boss_ready.execute(
"SELECT state, output FROM pgboss.job WHERE id=%s", (job["id"],)
).fetchone()
assert row[0] == "completed"
assert row[1] == {"pong": 1}
def test_fail_records_output_and_state(boss_ready):
boss_ready.execute("INSERT INTO pgboss.job(name) VALUES('echo')")
boss = Boss(dsn=__import__("void_workers.config", fromlist=["DATABASE_URL"]).DATABASE_URL)
job = boss.claim("echo")
boss.fail(job["id"], {"name": "BoomError", "message": "boom"})
row = boss_ready.execute(
"SELECT state, output, retry_count FROM pgboss.job WHERE id=%s",
(job["id"],)
).fetchone()
# retry_limit defaults to 5 and retry_count was 0 → should move to 'retry'
assert row[0] == "retry"
assert row[2] == 1
def test_fail_past_retry_limit_marks_failed(boss_ready):
boss_ready.execute(
"INSERT INTO pgboss.job(name, retry_count, retry_limit) VALUES('echo', 5, 5)"
)
boss = Boss(dsn=__import__("void_workers.config", fromlist=["DATABASE_URL"]).DATABASE_URL)
job = boss.claim("echo")
boss.fail(job["id"], {"name": "Done"})
row = boss_ready.execute(
"SELECT state FROM pgboss.job WHERE id=%s", (job["id"],)
).fetchone()
assert row[0] == "failed"
```
- [ ] **Step 3:** Run red:
```bash
cd /project/src/void-v2/workers
. .venv/bin/activate
DATABASE_URL="$(grep ^DATABASE_URL= ../.env 2>/dev/null | cut -d= -f2- ||
echo "postgres://void:$(ssh root@192.168.1.124 cat /root/void2-db-pass.txt | sed s/^DB_PASS=//)@192.168.1.215:5432/void")" \
pytest tests/test_boss.py -v
```
Expected: 4 failed (boss.py does not exist).
- [ ] **Step 4:** Implement `workers/void_workers/boss.py`:
```python
import json
import psycopg
from psycopg.rows import dict_row
class Boss:
def __init__(self, dsn):
self.dsn = dsn
def _conn(self):
return psycopg.connect(self.dsn, autocommit=False, row_factory=dict_row)
def claim(self, queue):
"""Atomically claim one job for the given queue. Returns dict or None."""
with self._conn() as conn:
with conn.cursor() as cur:
cur.execute("""
SELECT id, data, retry_count, retry_limit
FROM pgboss.job
WHERE name=%s
AND state IN ('created','retry')
AND start_after <= now()
ORDER BY priority DESC, created_on ASC
FOR UPDATE SKIP LOCKED
LIMIT 1
""", (queue,))
row = cur.fetchone()
if not row: return None
cur.execute(
"UPDATE pgboss.job SET state='active', started_on=now() WHERE id=%s",
(row["id"],)
)
conn.commit()
return row
def complete(self, job_id, output):
with self._conn() as conn:
with conn.cursor() as cur:
cur.execute("""
UPDATE pgboss.job
SET state='completed', completed_on=now(), output=%s
WHERE id=%s
""", (json.dumps(output), job_id))
conn.commit()
def fail(self, job_id, output):
"""Mark the job retry (if budget left) or failed (otherwise)."""
with self._conn() as conn:
with conn.cursor() as cur:
cur.execute(
"SELECT retry_count, retry_limit, retry_delay, retry_backoff "
"FROM pgboss.job WHERE id=%s FOR UPDATE",
(job_id,)
)
row = cur.fetchone()
if not row: return
new_count = row["retry_count"] + 1
if new_count > row["retry_limit"]:
cur.execute(
"UPDATE pgboss.job SET state='failed', output=%s, "
"completed_on=now() WHERE id=%s",
(json.dumps(output), job_id)
)
else:
# exponential backoff: delay * 2^(count-1)
delay = row["retry_delay"]
if row["retry_backoff"]:
delay = delay * (2 ** (new_count - 1))
cur.execute("""
UPDATE pgboss.job
SET state='retry',
retry_count=%s,
start_after=now() + (%s || ' seconds')::interval,
output=%s
WHERE id=%s
""", (new_count, str(delay), json.dumps(output), job_id))
conn.commit()
```
- [ ] **Step 5:** Run green:
```bash
DATABASE_URL=... pytest tests/test_boss.py -v
```
Expected: 4 passed.
- [ ] **Step 6: Commit.**
```bash
cd /project/src/void-v2
git add workers/void_workers/boss.py workers/tests/conftest.py workers/tests/test_boss.py
git commit -m "feat(workers): pgboss claim/complete/fail via psycopg"
```
### Task A3: Echo handler + runner loop
**Files:**
- Create: `workers/void_workers/handlers/__init__.py`
- Create: `workers/void_workers/handlers/echo.py`
- Create: `workers/void_workers/runner.py`
- Create: `workers/tests/test_echo.py`
- [ ] **Step 1:** Write failing test `workers/tests/test_echo.py`:
```python
import json
import threading
import time
from void_workers.boss import Boss
from void_workers.runner import Runner
from void_workers.config import DATABASE_URL
def test_echo_handler_runs(boss_ready):
boss_ready.execute(
"INSERT INTO pgboss.job(name, data) VALUES('echo', %s)",
(json.dumps({"ping": 42}),)
)
runner = Runner(dsn=DATABASE_URL, handlers={"echo": {"concurrency": 1}})
t = threading.Thread(target=runner.run, kwargs={"once": True}, daemon=True)
t.start()
t.join(timeout=5)
row = boss_ready.execute(
"SELECT state, output FROM pgboss.job WHERE name='echo' ORDER BY created_on DESC LIMIT 1"
).fetchone()
assert row[0] == "completed"
assert row[1] == {"pong": 42}
```
- [ ] **Step 2:** Run red:
```bash
DATABASE_URL=... pytest tests/test_echo.py -v
```
Expected: ImportError / 1 failed.
- [ ] **Step 3:** Implement `workers/void_workers/handlers/__init__.py`:
```python
from . import echo
REGISTRY = {
echo.NAME: echo.handle,
}
```
- [ ] **Step 4:** Implement `workers/void_workers/handlers/echo.py`:
```python
NAME = "echo"
def handle(job_data: dict) -> dict:
return {"pong": job_data.get("ping", 0)}
```
- [ ] **Step 5:** Implement `workers/void_workers/runner.py`:
```python
import signal
import threading
import time
from concurrent.futures import ThreadPoolExecutor
from .boss import Boss
from .config import CONCURRENCY, POLL_INTERVAL_MS, DATABASE_URL
from .handlers import REGISTRY
from .log import log
class Runner:
def __init__(self, dsn=None, handlers=None):
self.boss = Boss(dsn or DATABASE_URL)
self.handlers = handlers or {
name: {"concurrency": CONCURRENCY.get(name, 1)} for name in REGISTRY
}
self._stop = threading.Event()
def _process_one(self, queue):
job = self.boss.claim(queue)
if not job: return False
log.info("job_claimed", job_id=str(job["id"]), name=queue)
handler = REGISTRY[queue]
try:
out = handler(job["data"] or {})
self.boss.complete(job["id"], out or {})
log.info("job_completed", job_id=str(job["id"]), name=queue)
except Exception as e:
log.error("job_failed", job_id=str(job["id"]), name=queue, err=str(e))
self.boss.fail(job["id"], {"name": type(e).__name__, "message": str(e)})
return True
def _loop(self, queue):
interval = POLL_INTERVAL_MS / 1000.0
while not self._stop.is_set():
try:
did = self._process_one(queue)
except Exception as e:
log.error("loop_error", queue=queue, err=str(e))
did = False
if not did:
self._stop.wait(interval)
def run(self, once=False):
if once:
for q in self.handlers:
self._process_one(q)
return
executor = ThreadPoolExecutor(max_workers=sum(h["concurrency"] for h in self.handlers.values()))
for q, cfg in self.handlers.items():
for _ in range(cfg["concurrency"]):
executor.submit(self._loop, q)
for sig in (signal.SIGTERM, signal.SIGINT):
signal.signal(sig, lambda *_: self._stop.set())
while not self._stop.is_set():
time.sleep(0.5)
executor.shutdown(wait=True)
if __name__ == "__main__":
Runner().run()
```
- [ ] **Step 6:** Run green:
```bash
DATABASE_URL=... pytest tests/test_echo.py -v
```
Expected: 1 passed.
- [ ] **Step 7: Commit.**
```bash
cd /project/src/void-v2
git add workers/void_workers/handlers/__init__.py workers/void_workers/handlers/echo.py workers/void_workers/runner.py workers/tests/test_echo.py
git commit -m "feat(workers): runner loop + echo handler"
```
### Task A4: systemd unit + push-workers.sh
**Files:**
- Create: `deploy/void-workers.service`
- Create: `deploy/push-workers.sh`
- Modify: `deploy/README.md` (append workers bootstrap section)
- [ ] **Step 1:** Create `deploy/void-workers.service`:
```ini
[Unit]
Description=Void 2.0 workers
After=network-online.target void-server.service
Wants=network-online.target
[Service]
Type=simple
User=voidworkers
WorkingDirectory=/opt/void-workers
EnvironmentFile=/opt/void-workers/.env
ExecStart=/opt/void-workers/venv/bin/python -m void_workers.runner
Restart=on-failure
RestartSec=5
StandardOutput=journal
StandardError=journal
MemoryMax=6G
[Install]
WantedBy=multi-user.target
```
- [ ] **Step 2:** Create `deploy/push-workers.sh`:
```bash
#!/usr/bin/env bash
set -euo pipefail
# Push Python void-workers source to CT 311 and restart the service.
# Run from /project/src/void-v2.
TARGET=${TARGET:-root@192.168.1.13}
REMOTE_DIR=${REMOTE_DIR:-/opt/void-workers}
rsync -avz --delete \
--exclude .venv \
--exclude __pycache__ \
--exclude '*.egg-info' \
--exclude tests \
workers/ "$TARGET:$REMOTE_DIR/"
ssh "$TARGET" "
cd $REMOTE_DIR
if [ ! -d venv ]; then
python3.12 -m venv venv
fi
. venv/bin/activate
pip install --quiet --upgrade pip
pip install --quiet -e '.[all]'
systemctl restart void-workers
"
echo "Workers deployed."
```
Make it executable:
```bash
chmod +x deploy/push-workers.sh
```
- [ ] **Step 3:** Extend `deploy/README.md` — append the following section:
```markdown
## Workers (Python void-workers — Plan 4+)
Runs alongside void-server as a second systemd unit.
One-time setup on CT 311:
```bash
apt install -y python3.12 python3.12-venv python3-pip \
ffmpeg tesseract-ocr tesseract-ocr-eng poppler-utils
useradd -r -m -d /opt/void-workers -s /bin/bash voidworkers
mkdir -p /opt/void-workers /var/lib/void/whisper-models
chown voidworkers: /opt/void-workers
chown -R voidworkers: /var/lib/void/whisper-models
# voidworkers needs to read the shared blob store
usermod -aG void voidworkers
install -m 644 deploy/void-workers.service /etc/systemd/system/
systemctl daemon-reload
systemctl enable void-workers
# /opt/void-workers/.env — see plan §Deploy delta
```
Deploy after edits:
```bash
cd /project/src/void-v2
./deploy/push-workers.sh
```
```
- [ ] **Step 4:** Run the existing Node test suite — must still be green (we have not modified anything Node touches yet).
```bash
cd /project/src/void-v2
npx vitest run
```
- [ ] **Step 5: Commit.**
```bash
git add deploy/void-workers.service deploy/push-workers.sh deploy/README.md
git commit -m "feat(workers): systemd unit + push-workers.sh"
```
### Task A5: Phase A close — memory checkpoint
- [ ] **Step 1:** Update memory `project_void_v2_execution.md`: mark "Plan 4 Phase A complete: harness + echo running locally; void-workers.service installed on CT 311 pending real deps (Phase B+)".
- [ ] **Step 2:** Note the boss.py contract decision (psycopg + native SQL) so future plans don't re-litigate it.
---
## Phase B — `extract.pdf` + `extract.image`
### Task B1: Install system deps on the dev box
These deps support local pytest runs. The same packages will be installed on CT 311 when Phase A's systemd unit lands there.
- [ ] **Step 1:** Verify deps locally (we are running on Ubuntu 24.04 LXC per CLAUDE.md):
```bash
command -v pdftotext && command -v tesseract && command -v ffmpeg || echo "missing deps"
apt list --installed 2>/dev/null | grep -E "poppler|tesseract|ffmpeg" | head -5
```
- [ ] **Step 2:** If any are missing, install:
```bash
apt install -y poppler-utils tesseract-ocr tesseract-ocr-eng ffmpeg
```
- [ ] **Step 3:** Install pdf/image Python extras into the workers venv:
```bash
cd /project/src/void-v2/workers
. .venv/bin/activate
pip install -e '.[pdf,image,test]'
python -c "import pdfplumber, pytesseract; print('ok')"
```
- [ ] **Step 4:** Add a `tests/fixtures/` README explaining how fixtures were generated (so we can regenerate them):
Create `workers/tests/fixtures/README.md`:
```markdown
# Test fixtures
- `born_digital.pdf` — generated by
`pdftotext -version 2>&1 | head -1 > /tmp/b.txt && enscript -p /tmp/b.ps /tmp/b.txt && ps2pdf /tmp/b.ps born_digital.pdf`
(or any tool that produces a PDF with embedded text). Must contain
the string "void-workers".
- `scanned.pdf` — `convert -density 200 eng_text.png scanned.pdf` (PDF
that is a single page of an image — `pdftotext` will yield nothing,
Tesseract fallback must pick it up).
- `eng_text.png` — a PNG with a clear English sentence on white
background. Must contain the string "blackflame".
```
Generate them now:
```bash
cd /project/src/void-v2/workers/tests/fixtures
# eng_text.png: render text with ImageMagick (apt: imagemagick)
apt install -y imagemagick ghostscript
convert -size 800x200 xc:white -font DejaVu-Sans -pointsize 36 \
-fill black -annotate +50+100 "blackflame palette" eng_text.png
# born_digital.pdf: simple TXT → PDF
echo "void-workers proof-of-life document" > /tmp/b.txt
enscript -p /tmp/b.ps /tmp/b.txt 2>/dev/null || echo "enscript missing — using ImageMagick"
if [ -f /tmp/b.ps ]; then
ps2pdf /tmp/b.ps born_digital.pdf
else
convert -size 800x200 xc:white -font DejaVu-Sans -pointsize 24 \
-fill black -annotate +50+100 "void-workers proof-of-life" born_digital.pdf
# NOTE: this is a single-image PDF too — pdftotext will return empty.
# Override: use poppler's pdftohtml tools or create with text.
fi
# scanned.pdf: PNG → PDF
convert -density 200 eng_text.png scanned.pdf
ls -la
```
If `enscript` is not available, install or use another text-PDF tool. The key fixture invariant is that `pdftotext born_digital.pdf -` returns non-empty text.
- [ ] **Step 5: Commit fixtures.**
```bash
cd /project/src/void-v2
git add workers/tests/fixtures
git commit -m "test(workers): pdf/image test fixtures"
```
### Task B2: `extract.pdf` worker — pdftotext path
**Files:**
- Create: `workers/void_workers/handlers/pdf.py`
- Modify: `workers/void_workers/handlers/__init__.py`
- Create: `workers/void_workers/repo.py`
- Create: `workers/tests/test_pdf.py`
- [ ] **Step 1:** Write the failing test:
```python
# workers/tests/test_pdf.py
import json
import os
import psycopg
from pathlib import Path
from void_workers.handlers.pdf import handle as handle_pdf
from void_workers.config import DATABASE_URL
FIXTURES = Path(__file__).parent / "fixtures"
def _seed_space_and_ref(conn, blob_path, kind="pdf"):
"""Bring up the Void schema (matches migrate.js) just enough to insert a ref."""
sp = conn.execute(
"INSERT INTO spaces(slug, name) VALUES('plan4-tests', 'P4') "
"ON CONFLICT (slug) DO UPDATE SET name=EXCLUDED.name RETURNING id"
).fetchone()[0]
ref = conn.execute(
"INSERT INTO refs(space_id, kind, source_url, title, blob_path) "
"VALUES(%s, %s, NULL, 'fixture', %s) RETURNING id",
(sp, kind, str(blob_path))
).fetchone()[0]
return sp, ref
def test_pdf_born_digital_uses_pdftotext(conn):
"""A PDF with embedded text gets pdftotext-extracted body."""
conn.execute("DROP SCHEMA IF EXISTS public CASCADE; CREATE SCHEMA public")
# Apply Void migrations against the conn
_run_node_migrations()
blob = FIXTURES / "born_digital.pdf"
sp, ref = _seed_space_and_ref(conn, blob)
out = handle_pdf({"ref_id": str(ref), "blob_path": str(blob)})
assert out["ref_id"] == str(ref)
assert out["chars"] > 0
row = conn.execute("SELECT body_text FROM refs WHERE id=%s", (ref,)).fetchone()
assert "void-workers" in (row[0] or "").lower()
def _run_node_migrations():
"""Shell to Node's migrate runner — same DB."""
import subprocess
subprocess.run(
["node", "lib/db/migrate.js", "up"],
cwd="/project/src/void-v2",
check=True
)
```
- [ ] **Step 2:** Run red:
```bash
cd /project/src/void-v2/workers
. .venv/bin/activate
DATABASE_URL=... pytest tests/test_pdf.py -v
```
Expected: ImportError on pdf module.
- [ ] **Step 3:** Implement `workers/void_workers/repo.py`:
```python
import json
import psycopg
from psycopg.rows import dict_row
from .config import DATABASE_URL
def _conn():
return psycopg.connect(DATABASE_URL, autocommit=True, row_factory=dict_row)
def update_ref(ref_id, *, body_text=None, metadata_patch=None):
"""UPDATE refs ... + emit audit_log row with actor_kind='worker'."""
with _conn() as conn:
before = conn.execute("SELECT * FROM refs WHERE id=%s", (ref_id,)).fetchone()
sets, args = [], []
if body_text is not None:
sets.append("body_text=%s"); args.append(body_text)
if metadata_patch is not None:
sets.append("metadata = coalesce(metadata, '{}'::jsonb) || %s::jsonb")
args.append(json.dumps(metadata_patch))
if not sets: return before
sets.append("updated_at=now()")
args.append(ref_id)
after = conn.execute(
f"UPDATE refs SET {', '.join(sets)} WHERE id=%s RETURNING *",
args
).fetchone()
# audit log — match Node's audit.js diff shape (simple)
diff = {}
if body_text is not None and (before or {}).get("body_text") != body_text:
diff["body_text"] = {"before": "(redacted)", "after_len": len(body_text)}
conn.execute("""
INSERT INTO audit_log(actor_kind, actor_id, entity_type, entity_id, action, diff)
VALUES('worker', NULL, 'ref', %s, 'update', %s)
""", (ref_id, json.dumps({"kind": "update", "changes": diff})))
return after
```
- [ ] **Step 4:** Implement `workers/void_workers/handlers/pdf.py`:
```python
import subprocess
from .. import repo
NAME = "extract.pdf"
def _pdftotext(blob_path):
return subprocess.check_output(
["pdftotext", "-layout", blob_path, "-"],
timeout=120
).decode("utf-8", errors="replace")
def handle(job_data: dict) -> dict:
ref_id = job_data["ref_id"]
blob_path = job_data["blob_path"]
text = _pdftotext(blob_path).strip()
method = "pdftotext"
# Fallback (Tesseract OCR) lands in Task B3.
body_text = text[:200_000]
repo.update_ref(ref_id, body_text=body_text,
metadata_patch={"extract": {"method": method, "chars": len(body_text)}})
return {"ref_id": ref_id, "chars": len(body_text), "method": method}
```
- [ ] **Step 5:** Add `pdf.handle` to the registry — edit `workers/void_workers/handlers/__init__.py`:
```python
from . import echo, pdf
REGISTRY = {
echo.NAME: echo.handle,
pdf.NAME: pdf.handle,
}
```
- [ ] **Step 6:** Run green:
```bash
DATABASE_URL=... pytest tests/test_pdf.py::test_pdf_born_digital_uses_pdftotext -v
```
Expected: 1 passed.
- [ ] **Step 7: Commit.**
```bash
cd /project/src/void-v2
git add workers/void_workers/handlers/__init__.py workers/void_workers/handlers/pdf.py workers/void_workers/repo.py workers/tests/test_pdf.py
git commit -m "feat(workers): extract.pdf via pdftotext"
```
### Task B3: PDF Tesseract fallback for scanned pages
**Files:**
- Modify: `workers/void_workers/handlers/pdf.py`
- Modify: `workers/tests/test_pdf.py`
- [ ] **Step 1:** Add the failing test (append to `test_pdf.py`):
```python
def test_pdf_scanned_falls_back_to_tesseract(conn):
conn.execute("DROP SCHEMA IF EXISTS public CASCADE; CREATE SCHEMA public")
_run_node_migrations()
blob = FIXTURES / "scanned.pdf"
sp, ref = _seed_space_and_ref(conn, blob)
out = handle_pdf({"ref_id": str(ref), "blob_path": str(blob)})
assert out["method"] == "tesseract"
row = conn.execute("SELECT body_text FROM refs WHERE id=%s", (ref,)).fetchone()
assert "blackflame" in (row[0] or "").lower()
```
- [ ] **Step 2:** Run red.
- [ ] **Step 3:** Implement the fallback in `workers/void_workers/handlers/pdf.py`:
```python
import io
import subprocess
import tempfile
from pathlib import Path
from PIL import Image
import pytesseract
from .. import repo
NAME = "extract.pdf"
FALLBACK_THRESHOLD = 200 # chars
def _pdftotext(blob_path):
return subprocess.check_output(
["pdftotext", "-layout", blob_path, "-"], timeout=120
).decode("utf-8", errors="replace")
def _ocr_pdf(blob_path):
"""Rasterize each page with pdftoppm, OCR each with Tesseract."""
with tempfile.TemporaryDirectory() as tmp:
subprocess.run(
["pdftoppm", "-r", "200", "-png", blob_path, f"{tmp}/p"],
check=True, timeout=300
)
pages = sorted(Path(tmp).glob("p-*.png"))
parts = []
for p in pages:
img = Image.open(p)
parts.append(pytesseract.image_to_string(img, lang="eng"))
return "\n".join(parts)
def handle(job_data: dict) -> dict:
ref_id = job_data["ref_id"]
blob_path = job_data["blob_path"]
method = "pdftotext"
text = _pdftotext(blob_path).strip()
if len(text) < FALLBACK_THRESHOLD:
method = "tesseract"
text = _ocr_pdf(blob_path).strip()
body_text = text[:200_000]
repo.update_ref(ref_id, body_text=body_text,
metadata_patch={"extract": {"method": method, "chars": len(body_text)}})
return {"ref_id": ref_id, "chars": len(body_text), "method": method}
```
- [ ] **Step 4:** Run green:
```bash
DATABASE_URL=... pytest tests/test_pdf.py -v
```
Expected: 2 passed.
- [ ] **Step 5: Commit.**
```bash
git add workers/void_workers/handlers/pdf.py workers/tests/test_pdf.py
git commit -m "feat(workers): pdf Tesseract fallback for scanned pages"
```
### Task B4: `extract.image` worker
**Files:**
- Create: `workers/void_workers/handlers/image.py`
- Modify: `workers/void_workers/handlers/__init__.py`
- Create: `workers/tests/test_image.py`
- [ ] **Step 1:** Write the failing test `workers/tests/test_image.py`:
```python
import os
from pathlib import Path
from void_workers.handlers.image import handle as handle_image
FIXTURES = Path(__file__).parent / "fixtures"
def test_image_ocr(conn):
conn.execute("DROP SCHEMA IF EXISTS public CASCADE; CREATE SCHEMA public")
import subprocess
subprocess.run(["node", "lib/db/migrate.js", "up"], cwd="/project/src/void-v2", check=True)
sp = conn.execute(
"INSERT INTO spaces(slug, name) VALUES('plan4-img', 'I') RETURNING id"
).fetchone()[0]
blob = FIXTURES / "eng_text.png"
ref = conn.execute(
"INSERT INTO refs(space_id, kind, blob_path) VALUES(%s, 'image', %s) RETURNING id",
(sp, str(blob))
).fetchone()[0]
out = handle_image({"ref_id": str(ref), "blob_path": str(blob)})
assert out["chars"] > 0
row = conn.execute("SELECT body_text FROM refs WHERE id=%s", (ref,)).fetchone()
assert "blackflame" in (row[0] or "").lower()
```
- [ ] **Step 2:** Run red.
- [ ] **Step 3:** Implement `workers/void_workers/handlers/image.py`:
```python
from PIL import Image
import pytesseract
from .. import repo
NAME = "extract.image"
def handle(job_data: dict) -> dict:
ref_id = job_data["ref_id"]
blob_path = job_data["blob_path"]
text = pytesseract.image_to_string(Image.open(blob_path), lang="eng").strip()
body_text = text[:200_000]
repo.update_ref(ref_id, body_text=body_text,
metadata_patch={"extract": {"method": "tesseract", "chars": len(body_text)}})
return {"ref_id": ref_id, "chars": len(body_text)}
```
- [ ] **Step 4:** Register — edit `workers/void_workers/handlers/__init__.py`:
```python
from . import echo, pdf, image
REGISTRY = {
echo.NAME: echo.handle,
pdf.NAME: pdf.handle,
image.NAME: image.handle,
}
```
- [ ] **Step 5:** Run green.
- [ ] **Step 6: Commit.**
```bash
git add workers/void_workers/handlers/__init__.py workers/void_workers/handlers/image.py workers/tests/test_image.py
git commit -m "feat(workers): extract.image via Tesseract"
```
### Task B5: Node side — blob worker enqueues extract jobs
**Files:**
- Modify: `lib/jobs/workers/blob.js`
- Modify: `tests/jobs/workers/blob.test.js`
- [ ] **Step 1:** Extend the existing blob.test.js with a new assertion. Append:
```js
it('enqueues extract.pdf after creating a pdf ref', async () => {
const sp = await spaces.create({ slug: 'bpdf', name: 'BPdf' }, { kind: 'user', id: null });
const upTmp = path.join(tmpRoot, 'doc.tmp');
await fs.writeFile(upTmp, Buffer.from('%PDF-1.4 ...'));
const id = await queue.enqueue('ingest.blob', {
space_id: sp.id, tmp_path: upTmp, filename: 'a.pdf', content_type: 'application/pdf'
});
await waitForJob('ingest.blob', id, { timeoutMs: 10_000 });
const { default: jobsRepo } = await import('../../../lib/db/repos/jobs.js');
const followUp = (await jobsRepo.list({ name: 'extract.pdf' })).find(j => j.data?.blob_path);
expect(followUp).toBeTruthy();
});
it('enqueues extract.image after creating an image ref', async () => {
const sp = await spaces.create({ slug: 'bimg', name: 'BImg' }, { kind: 'user', id: null });
const upTmp = path.join(tmpRoot, 'pic.tmp');
await fs.writeFile(upTmp, Buffer.from([0x89, 0x50, 0x4e, 0x47]));
const id = await queue.enqueue('ingest.blob', {
space_id: sp.id, tmp_path: upTmp, filename: 'a.png', content_type: 'image/png'
});
await waitForJob('ingest.blob', id, { timeoutMs: 10_000 });
const { default: jobsRepo } = await import('../../../lib/db/repos/jobs.js');
const followUp = (await jobsRepo.list({ name: 'extract.image' })).find(j => j.data?.blob_path);
expect(followUp).toBeTruthy();
});
```
- [ ] **Step 2:** Run red:
```bash
cd /project/src/void-v2
npx vitest run tests/jobs/workers/blob.test.js
```
Expected: 2 new tests fail (no extract.* enqueue yet).
- [ ] **Step 3:** Modify `lib/jobs/workers/blob.js` to enqueue the follow-up job:
```js
// existing imports
import * as queue from '../queue.js';
// ... inside handler, after `const row = await refs.create(...)`:
if (kind === 'pdf') {
await queue.enqueue('extract.pdf', { ref_id: row.id, blob_path: path });
} else if (kind === 'image') {
await queue.enqueue('extract.image', { ref_id: row.id, blob_path: path });
}
return { ref_id: row.id, sha };
```
(Apply the exact edit at the `return { ref_id: row.id, sha };` site.)
- [ ] **Step 4:** Run green:
```bash
npx vitest run tests/jobs/workers/blob.test.js
```
Expected: 4 passed.
- [ ] **Step 5:** Run the full suite — must still be green.
```bash
npx vitest run
```
- [ ] **Step 6: Commit.**
```bash
git add lib/jobs/workers/blob.js tests/jobs/workers/blob.test.js
git commit -m "feat(jobs): blob worker fans out to extract.pdf / extract.image"
```
### Task B6: Phase B close — memory + full suite
- [ ] **Step 1:** `npx vitest run` — full Node suite green.
- [ ] **Step 2:** `cd workers && DATABASE_URL=... pytest -v` — full Python suite green.
- [ ] **Step 3:** Update memory: "Plan 4 Phase B complete: extract.pdf + extract.image workers; Node blob worker fans out."
---
## Phase C — `ingest.video` + CT 311 resize + GPU passthrough
### Task C1: Snapshot + CT 311 resize + GPU passthrough
User pre-authorized this 2026-06-01.
- [ ] **Step 1:** Snapshot CT 310 + CT 311.
```bash
SNAP=plan4_pre_resize_$(date +%Y%m%d_%H%M)
ssh root@192.168.1.124 "pct snapshot 310 $SNAP --description 'Pre CT 311 resize + GPU passthrough'" &
ssh root@192.168.1.124 "pct snapshot 311 $SNAP --description 'Pre CT 311 resize + GPU passthrough'" &
wait
ssh root@192.168.1.124 "pct listsnapshot 310 | tail -3; echo ---; pct listsnapshot 311 | tail -3"
```
- [ ] **Step 2:** Resize CT 311.
```bash
ssh root@192.168.1.124 "pct set 311 -memory 8192 -cores 6"
ssh root@192.168.1.124 "pct config 311 | grep -E 'cores|memory'"
```
- [ ] **Step 3:** Add GPU device-node passthrough lines (copy from CT 102).
```bash
ssh root@192.168.1.124 "
pct set 311 -dev0 /dev/nvidia0,gid=44
pct set 311 -dev1 /dev/nvidiactl,gid=44
pct set 311 -dev2 /dev/nvidia-uvm,gid=44
pct set 311 -dev3 /dev/nvidia-uvm-tools,gid=44
pct set 311 -dev4 /dev/nvidia-caps/nvidia-cap1,gid=44
pct set 311 -dev5 /dev/nvidia-caps/nvidia-cap2,gid=44
pct config 311 | grep dev
"
```
- [ ] **Step 4:** Restart CT 311.
```bash
ssh root@192.168.1.124 "pct reboot 311"
sleep 10
ssh root@192.168.1.13 "ls -la /dev/nvidia* 2>&1 | head; nvidia-smi --query-gpu=name --format=csv,noheader 2>&1 || echo 'nvidia-smi missing (no driver in CT) — expected'"
curl -s http://192.168.1.13:3000/health
```
Expected: device nodes visible, health returns alpha-3.
- [ ] **Step 5: Commit (config-as-comment).** No code change yet — we record the infra step in the next phase's commit message.
### Task C2: Install ffmpeg + faster-whisper deps on CT 311
- [ ] **Step 1:** Install system deps:
```bash
ssh root@192.168.1.13 "
apt update
apt install -y python3.12 python3.12-venv python3-pip \
ffmpeg tesseract-ocr tesseract-ocr-eng poppler-utils \
nvidia-utils-535 nvidia-cuda-toolkit 2>&1 | tail -5
python3.12 --version
ffmpeg -version | head -1
tesseract --version 2>&1 | head -1
"
```
Note: `nvidia-cuda-toolkit` is for `ctranslate2`'s CUDA runtime; if you'd rather not pull the full toolkit (~1 GB), use the official `libcudnn8 libcublas12` runtime packages instead.
- [ ] **Step 2:** Create the voidworkers user + dirs on CT 311:
```bash
ssh root@192.168.1.13 "
id voidworkers >/dev/null 2>&1 || useradd -r -m -d /opt/void-workers -s /bin/bash voidworkers
mkdir -p /opt/void-workers /var/lib/void/whisper-models
chown voidworkers: /opt/void-workers
chown -R voidworkers: /var/lib/void/whisper-models
usermod -aG void voidworkers
chmod -R g+rX /var/lib/void/blobs # ensure voidworkers (in void group) can read blobs
install -m 644 /opt/void-workers/.env 2>/dev/null || true
cat > /opt/void-workers/.env <<EOF
DATABASE_URL=$(grep ^DATABASE_URL= /opt/void-server/.env | cut -d= -f2-)
BLOB_ROOT=/var/lib/void/blobs
WHISPER_MODEL=small.en
WHISPER_CACHE=/var/lib/void/whisper-models
EOF
chmod 600 /opt/void-workers/.env
chown voidworkers: /opt/void-workers/.env
"
```
- [ ] **Step 3:** Copy the systemd unit:
```bash
scp /project/src/void-v2/deploy/void-workers.service root@192.168.1.13:/etc/systemd/system/void-workers.service
ssh root@192.168.1.13 "systemctl daemon-reload && systemctl enable void-workers"
```
(Don't start it yet — venv is not installed; `push-workers.sh` will set it up.)
- [ ] **Step 4: Initial push.**
```bash
cd /project/src/void-v2
./deploy/push-workers.sh
ssh root@192.168.1.13 "systemctl status void-workers --no-pager -n 5"
```
Expected: active (running), waiting for jobs.
- [ ] **Step 5:** Sanity — enqueue an echo via Node, observe Python claim it.
```bash
TOKEN=$(ssh root@192.168.1.13 "grep ^OWNER_TOKEN= /opt/void-server/.env | cut -d= -f2-")
# Manually insert an echo via psql since /api/capture doesn't enqueue 'echo':
DBPASS=$(ssh root@192.168.1.124 'cat /root/void2-db-pass.txt' | tr -d '\r\n' | sed 's/^DB_PASS=//')
PGPASSWORD="$DBPASS" psql -h 192.168.1.215 -U void -d void -c "INSERT INTO pgboss.job(name, data) VALUES('echo', '{\"ping\": 99}'::jsonb) RETURNING id"
sleep 3
PGPASSWORD="$DBPASS" psql -h 192.168.1.215 -U void -d void -c "SELECT state, output FROM pgboss.job WHERE name='echo' ORDER BY created_on DESC LIMIT 1"
```
Expected: `completed | {"pong": 99}`.
- [ ] **Step 6: Commit deploy-side changes (none in repo from this task, but mark the milestone).**
No commit — this task is operational on CT 311. Note in memory.
### Task C3: Whisper model loader (`model.py`)
**Files:**
- Create: `workers/void_workers/model.py`
- Create: `workers/tests/test_model.py`
- [ ] **Step 1:** Failing test (uses mock):
```python
# workers/tests/test_model.py
from unittest.mock import patch, MagicMock
from void_workers import model
def test_model_returns_singleton(monkeypatch):
m = MagicMock()
monkeypatch.setattr(model, "_whisper_model", None)
with patch("void_workers.model.cuda_available", return_value=False):
with patch("faster_whisper.WhisperModel", return_value=m):
a = model.whisper_model()
b = model.whisper_model()
assert a is b
def test_transcribe_returns_joined_segments(monkeypatch):
seg1 = MagicMock(text=" Hello world ")
seg2 = MagicMock(text=" second line")
fake_model = MagicMock()
fake_model.transcribe.return_value = ([seg1, seg2], MagicMock())
monkeypatch.setattr(model, "_whisper_model", fake_model)
out = model.whisper_transcribe("/tmp/whatever.opus")
assert "Hello world" in out
assert "second line" in out
```
- [ ] **Step 2:** Run red.
- [ ] **Step 3:** Implement `workers/void_workers/model.py`:
```python
import os
from .log import log
_whisper_model = None
def cuda_available():
try:
import ctranslate2
return ctranslate2.get_cuda_device_count() > 0
except Exception as e:
log.info("ctranslate2_cuda_probe_failed", err=str(e))
return False
def whisper_model():
global _whisper_model
if _whisper_model is None:
from faster_whisper import WhisperModel
name = os.environ.get("WHISPER_MODEL", "small.en")
cache = os.environ.get("WHISPER_CACHE", "/var/lib/void/whisper-models")
device = "cuda" if cuda_available() else "cpu"
compute_type = "float16" if device == "cuda" else "int8"
log.info("whisper_loading", model=name, device=device, compute_type=compute_type, cache=cache)
_whisper_model = WhisperModel(name, device=device, compute_type=compute_type, download_root=cache)
return _whisper_model
def whisper_transcribe(audio_path):
segments, _info = whisper_model().transcribe(audio_path, vad_filter=True)
return "\n".join(s.text.strip() for s in segments).strip()
```
- [ ] **Step 4:** Run green:
```bash
pytest tests/test_model.py -v
```
Expected: 2 passed.
- [ ] **Step 5: Commit.**
```bash
git add workers/void_workers/model.py workers/tests/test_model.py
git commit -m "feat(workers): whisper loader with CUDA detect + CPU fallback"
```
### Task C4: `ingest.video` worker (yt-dlp + transcribe + create ref)
**Files:**
- Create: `workers/void_workers/handlers/video.py`
- Modify: `workers/void_workers/handlers/__init__.py`
- Create: `workers/tests/test_video.py`
- Modify: `workers/void_workers/repo.py` — add `create_ref()`
- [ ] **Step 1:** Failing test (mocked yt-dlp + mocked Whisper):
```python
# workers/tests/test_video.py
from unittest.mock import patch, MagicMock
from void_workers.handlers.video import handle as handle_video
def test_video_creates_ref_with_transcript_and_metadata(conn, monkeypatch):
conn.execute("DROP SCHEMA IF EXISTS public CASCADE; CREATE SCHEMA public")
import subprocess
subprocess.run(["node", "lib/db/migrate.js", "up"], cwd="/project/src/void-v2", check=True)
sp = conn.execute("INSERT INTO spaces(slug, name) VALUES('plan4-vid', 'V') RETURNING id").fetchone()[0]
info = {
"title": "Sample video",
"description": "a description",
"duration": 90,
"uploader": "Channel",
"thumbnail": "https://i.ytimg.com/t.jpg"
}
with patch("void_workers.handlers.video._yt_dlp_info", return_value=info), \
patch("void_workers.handlers.video._yt_dlp_audio", return_value="/tmp/fake.opus"), \
patch("void_workers.handlers.video.whisper_transcribe", return_value="hello world transcript"), \
patch("os.unlink"):
out = handle_video({"space_id": str(sp), "url": "https://youtu.be/abc"})
assert "ref_id" in out
row = conn.execute("SELECT title, body_text, source_kind FROM refs WHERE id=%s", (out["ref_id"],)).fetchone()
assert row[0] == "Sample video"
assert "hello world" in row[1]
assert row[2] == "youtube"
```
- [ ] **Step 2:** Run red.
- [ ] **Step 3:** Implement `workers/void_workers/handlers/video.py`:
```python
import hashlib
import os
import subprocess
import tempfile
import json
from .. import repo
from ..model import whisper_transcribe
NAME = "ingest.video"
def _yt_dlp_info(url):
"""Returns dict of metadata, or None if yt-dlp could not extract."""
try:
out = subprocess.check_output(
["yt-dlp", "-J", "--no-warnings", "--no-playlist", url],
timeout=60
)
return json.loads(out)
except subprocess.CalledProcessError as e:
return None
def _yt_dlp_audio(url):
"""Downloads bestaudio to a temp .opus and returns the path."""
tmp_dir = tempfile.mkdtemp(prefix="void-yt-")
out_template = os.path.join(tmp_dir, "audio.%(ext)s")
subprocess.run(
["yt-dlp", "-x", "--audio-format", "opus", "-o", out_template,
"--no-warnings", "--no-playlist", url],
check=True, timeout=600
)
for f in os.listdir(tmp_dir):
if f.startswith("audio."):
return os.path.join(tmp_dir, f)
raise RuntimeError("yt-dlp produced no audio file")
def _idem(space_id, url):
return hashlib.sha256((space_id + "\x00" + url).encode()).hexdigest()
def _kind(url):
return "youtube" if ("youtube.com" in url or "youtu.be" in url) else "video"
def handle(job_data: dict) -> dict:
space_id = job_data["space_id"]
url = job_data["url"]
idem = _idem(space_id, url)
info = _yt_dlp_info(url)
if info is None:
return {"skipped": "yt-dlp"}
audio_path = _yt_dlp_audio(url)
try:
transcript = whisper_transcribe(audio_path)
finally:
try: os.unlink(audio_path)
except OSError: pass
ref_id = repo.create_ref({
"space_id": space_id,
"kind": "video",
"source_url": url,
"title": info.get("title") or url,
"summary": (info.get("description") or "")[:5000],
"body_text": transcript[:200_000],
"source_kind": _kind(url),
"external_id": idem,
"metadata": {
"duration_s": info.get("duration"),
"uploader": info.get("uploader"),
"thumbnail": info.get("thumbnail"),
"extract": {"method": "whisper", "chars": len(transcript)},
}
})
return {"ref_id": ref_id, "chars": len(transcript)}
```
- [ ] **Step 4:** Extend `workers/void_workers/repo.py` with `create_ref`:
```python
def create_ref(input_):
"""INSERT INTO refs(...) RETURNING id; emit audit_log."""
fields = ["space_id","kind","source_url","title","summary","body_text",
"blob_path","metadata","source_kind","external_id","captured_at"]
cols, vals = [], []
for f in fields:
if f in input_:
cols.append(f); vals.append(input_[f])
placeholders = ",".join(f"%s" for _ in cols)
sql = f"INSERT INTO refs({','.join(cols)}) VALUES({placeholders}) RETURNING id"
with _conn() as conn:
ref_id = conn.execute(sql, vals).fetchone()["id"]
conn.execute("""
INSERT INTO audit_log(actor_kind, actor_id, entity_type, entity_id, action, diff)
VALUES('worker', NULL, 'ref', %s, 'create', %s)
""", (ref_id, json.dumps({"kind": "create"})))
return str(ref_id)
```
(Make sure `json` import is at the top of `repo.py` if not already there. Also note `metadata` must be `json.dumps`'d before insertion because we're passing a dict; tweak the field handling to JSON-encode dict/list values.)
Patch `create_ref` to JSON-encode jsonb fields:
```python
def _maybe_json(v):
return json.dumps(v) if isinstance(v, (dict, list)) else v
def create_ref(input_):
fields = ["space_id","kind","source_url","title","summary","body_text",
"blob_path","metadata","source_kind","external_id","captured_at"]
cols, vals = [], []
for f in fields:
if f in input_:
cols.append(f); vals.append(_maybe_json(input_[f]))
placeholders = ",".join(f"%s" for _ in cols)
sql = f"INSERT INTO refs({','.join(cols)}) VALUES({placeholders}) RETURNING id"
with _conn() as conn:
ref_id = conn.execute(sql, vals).fetchone()["id"]
conn.execute("""
INSERT INTO audit_log(actor_kind, actor_id, entity_type, entity_id, action, diff)
VALUES('worker', NULL, 'ref', %s, 'create', %s)
""", (ref_id, json.dumps({"kind": "create"})))
return str(ref_id)
```
- [ ] **Step 5:** Register — edit `workers/void_workers/handlers/__init__.py`:
```python
from . import echo, pdf, image, video
REGISTRY = {
echo.NAME: echo.handle,
pdf.NAME: pdf.handle,
image.NAME: image.handle,
video.NAME: video.handle,
}
```
- [ ] **Step 6:** Run green:
```bash
pytest tests/test_video.py -v
```
Expected: 1 passed.
- [ ] **Step 7: Commit.**
```bash
git add workers/void_workers/handlers/__init__.py workers/void_workers/handlers/video.py workers/void_workers/repo.py workers/tests/test_video.py
git commit -m "feat(workers): ingest.video via yt-dlp + Whisper"
```
### Task C5: Node — capture.js routes YouTube to `ingest.video`
**Files:**
- Modify: `lib/api/routes/capture.js`
- Modify: `tests/api/capture.test.js`
- [ ] **Step 1:** Failing test — append to `tests/api/capture.test.js`:
```js
it('POST /api/capture with YouTube URL enqueues ingest.video, not ingest.url', async () => {
const res = await request(app).post('/api/capture').set(ownerHeaders)
.send({ space_id: sp.id, url: 'https://youtu.be/abc' });
expect(res.status).toBe(202);
expect(res.body.job_id).toBeTruthy();
const { default: jobsRepo } = await import('../../lib/db/repos/jobs.js');
const rows = await jobsRepo.list({ name: 'ingest.video' });
expect(rows.find(r => r.id === res.body.job_id)).toBeTruthy();
const urlRows = await jobsRepo.list({ name: 'ingest.url' });
expect(urlRows.find(r => r.id === res.body.job_id)).toBeFalsy();
});
```
- [ ] **Step 2:** Run red.
- [ ] **Step 3:** Modify `lib/api/routes/capture.js` — replace the `POST /` handler's enqueue logic:
```js
// helper near the top
const VIDEO_HOST_RE = /(^|\.)(youtube\.com|youtu\.be|vimeo\.com)$/i;
function isVideoUrl(url) {
try { return VIDEO_HOST_RE.test(new URL(url).hostname); }
catch { return false; }
}
// inside router.post('/'), after computing idem + the existing-ref check:
const isVideo = isVideoUrl(url);
const job_name = isVideo ? 'ingest.video' : 'ingest.url';
const job_id = await queue.enqueue(job_name, { space_id, url });
res.status(202).json({ job_id, idempotency_key: idem });
```
- [ ] **Step 4:** Run green:
```bash
npx vitest run tests/api/capture.test.js
```
Expected: existing + new tests pass.
- [ ] **Step 5: Commit.**
```bash
git add lib/api/routes/capture.js tests/api/capture.test.js
git commit -m "feat(api): capture routes YouTube/Vimeo URLs to ingest.video"
```
### Task C6: Phase C close — snapshot + memory
- [ ] **Step 1:** Run full Node + Python test suites green.
```bash
cd /project/src/void-v2 && npx vitest run
cd workers && DATABASE_URL=... pytest -v
```
- [ ] **Step 2:** Snapshot CT 310 + 311 — `plan4_phase_c_<timestamp>`.
- [ ] **Step 3:** Update memory: "Plan 4 Phase C complete: Whisper + yt-dlp live; CT 311 resized 6c/8G + GPU passthrough; voidworkers.service running on CT 311."
---
## Phase D — Source-doc sync + version bump + completion doc
### Task D1: Python `safe_fetch` port
**Files:**
- Create: `workers/void_workers/safe_fetch.py`
- Create: `workers/tests/test_safe_fetch.py`
- [ ] **Step 1:** Failing test:
```python
# workers/tests/test_safe_fetch.py
import pytest
from void_workers.safe_fetch import safe_fetch, SafeFetchError
def test_rejects_file_scheme():
with pytest.raises(SafeFetchError):
safe_fetch("file:///etc/passwd")
def test_rejects_loopback():
with pytest.raises(SafeFetchError):
safe_fetch("http://127.0.0.1/x")
def test_rejects_rfc1918():
with pytest.raises(SafeFetchError):
safe_fetch("http://192.168.1.1/x")
def test_rejects_metadata_endpoint():
with pytest.raises(SafeFetchError):
safe_fetch("http://169.254.169.254/latest/")
```
- [ ] **Step 2:** Run red.
- [ ] **Step 3:** Implement `workers/void_workers/safe_fetch.py`:
```python
import socket
import ipaddress
import urllib.request
import os
from urllib.parse import urlparse
BLOCK_V4_NETS = [ipaddress.ip_network(c) for c in [
"0.0.0.0/8", "127.0.0.0/8", "10.0.0.0/8",
"172.16.0.0/12", "192.168.0.0/16",
"169.254.0.0/16", "100.64.0.0/10",
]]
class SafeFetchError(Exception):
pass
def _is_blocked(addr):
if os.environ.get("VOID_INGEST_ALLOW_PRIVATE") == "true":
return False
try:
ip = ipaddress.ip_address(addr)
except ValueError:
return True
if ip.is_loopback or ip.is_link_local or ip.is_multicast or ip.is_unspecified:
return True
if isinstance(ip, ipaddress.IPv4Address):
return any(ip in n for n in BLOCK_V4_NETS)
# IPv6: ULA + link-local
if ip in ipaddress.ip_network("fc00::/7") or ip in ipaddress.ip_network("fe80::/10"):
return True
return False
def _resolve(host):
try:
infos = socket.getaddrinfo(host, None)
except socket.gaierror as e:
raise SafeFetchError(f"no DNS for {host}: {e}")
addrs = list({i[4][0] for i in infos})
for a in addrs:
if _is_blocked(a):
raise SafeFetchError(f"{host} resolves to blocked address {a}")
if not addrs:
raise SafeFetchError(f"no addresses for {host}")
return addrs[0]
def safe_fetch(url, *, headers=None, timeout=15, max_hops=5):
current = url
for hop in range(max_hops + 1):
u = urlparse(current)
if u.scheme not in ("http", "https"):
raise SafeFetchError(f"unsupported scheme {u.scheme}")
host = u.hostname
# If host is a literal IP, validate; else resolve + validate
try:
ipaddress.ip_address(host)
if _is_blocked(host):
raise SafeFetchError(f"blocked literal IP {host}")
except ValueError:
_resolve(host)
req = urllib.request.Request(current, headers=headers or {})
try:
opener = urllib.request.build_opener(urllib.request.HTTPRedirectHandler)
with opener.open(req, timeout=timeout) as r:
return r.read()
except urllib.error.HTTPError as e:
if e.code in (301, 302, 303, 307, 308) and "Location" in e.headers and hop < max_hops:
current = e.headers["Location"]
continue
raise
raise SafeFetchError(f"too many redirects ({max_hops})")
```
- [ ] **Step 4:** Run green.
- [ ] **Step 5: Commit.**
```bash
git add workers/void_workers/safe_fetch.py workers/tests/test_safe_fetch.py
git commit -m "feat(workers): safe_fetch (Python port)"
```
### Task D2: `sync.source_doc` worker
**Files:**
- Create: `workers/void_workers/handlers/sourcedoc.py`
- Modify: `workers/void_workers/handlers/__init__.py`
- Create: `workers/tests/test_sourcedoc.py`
- Modify: `workers/void_workers/repo.py` — add `get_source_doc`, `update_source_doc`
- [ ] **Step 1:** Failing test:
```python
# workers/tests/test_sourcedoc.py
from unittest.mock import patch
from void_workers.handlers.sourcedoc import handle as handle_sd
def test_sourcedoc_updates_body_text(conn):
conn.execute("DROP SCHEMA IF EXISTS public CASCADE; CREATE SCHEMA public")
import subprocess
subprocess.run(["node", "lib/db/migrate.js", "up"], cwd="/project/src/void-v2", check=True)
sp = conn.execute("INSERT INTO spaces(slug, name) VALUES('sd', 'SD') RETURNING id").fetchone()[0]
res = conn.execute(
"INSERT INTO resources(space_id, slug, name, runtime_type) "
"VALUES(%s, 'r', 'R', 'lxc') RETURNING id", (sp,)
).fetchone()[0]
sd = conn.execute(
"INSERT INTO source_docs(resource_id, name, upstream_url, sync_source) "
"VALUES(%s, 'doc', 'https://example.com/r', 'url') RETURNING id",
(res,)
).fetchone()[0]
with patch("void_workers.handlers.sourcedoc.safe_fetch",
return_value=b"hello world doc body"):
out = handle_sd({"source_doc_id": str(sd)})
assert out.get("updated") is True
row = conn.execute("SELECT body_text FROM source_docs WHERE id=%s", (sd,)).fetchone()
assert "hello world" in (row[0] or "")
```
- [ ] **Step 2:** Run red.
- [ ] **Step 3:** Extend `workers/void_workers/repo.py` with the two helpers:
```python
def get_source_doc(source_doc_id):
with _conn() as conn:
return conn.execute(
"SELECT * FROM source_docs WHERE id=%s", (source_doc_id,)
).fetchone()
def update_source_doc(source_doc_id, *, body_text=None, last_synced=None, metadata_patch=None):
with _conn() as conn:
before = conn.execute(
"SELECT * FROM source_docs WHERE id=%s", (source_doc_id,)
).fetchone()
sets, args = [], []
if body_text is not None:
sets.append("body_text=%s"); args.append(body_text)
if last_synced is not None:
sets.append("last_synced=%s"); args.append(last_synced)
if metadata_patch is not None:
sets.append("metadata = coalesce(metadata, '{}'::jsonb) || %s::jsonb")
args.append(json.dumps(metadata_patch))
if not sets: return before
sets.append("updated_at=now()")
args.append(source_doc_id)
after = conn.execute(
f"UPDATE source_docs SET {', '.join(sets)} WHERE id=%s RETURNING *", args
).fetchone()
conn.execute("""
INSERT INTO audit_log(actor_kind, actor_id, entity_type, entity_id, action, diff)
VALUES('worker', NULL, 'source_doc', %s, 'update', %s)
""", (source_doc_id, json.dumps({"kind":"update","changes":{"body_text":"updated" if body_text is not None else None}})))
return after
```
- [ ] **Step 4:** Implement `workers/void_workers/handlers/sourcedoc.py`:
```python
import hashlib
from datetime import datetime, timezone
from .. import repo
from ..safe_fetch import safe_fetch
NAME = "sync.source_doc"
def _sha(data):
return hashlib.sha256(data).hexdigest()
def handle(job_data: dict) -> dict:
sd_id = job_data["source_doc_id"]
doc = repo.get_source_doc(sd_id)
if not doc:
return {"skipped": "gone"}
body = safe_fetch(doc["upstream_url"], headers={"User-Agent": "void-ingest/2.0"})
new_sha = _sha(body)
old_sha = ((doc.get("metadata") or {}).get("body_sha"))
now = datetime.now(timezone.utc)
if new_sha == old_sha:
repo.update_source_doc(sd_id, last_synced=now)
return {"unchanged": True}
text = body.decode("utf-8", errors="replace")[:1_000_000]
repo.update_source_doc(
sd_id,
body_text=text,
last_synced=now,
metadata_patch={"body_sha": new_sha}
)
return {"updated": True, "chars": len(text)}
```
- [ ] **Step 5:** Register — edit `workers/void_workers/handlers/__init__.py`:
```python
from . import echo, pdf, image, video, sourcedoc
REGISTRY = {
echo.NAME: echo.handle,
pdf.NAME: pdf.handle,
image.NAME: image.handle,
video.NAME: video.handle,
sourcedoc.NAME: sourcedoc.handle,
}
```
- [ ] **Step 6:** Run green.
- [ ] **Step 7: Commit.**
```bash
git add workers/void_workers/handlers/__init__.py workers/void_workers/handlers/sourcedoc.py workers/void_workers/repo.py workers/tests/test_sourcedoc.py
git commit -m "feat(workers): sync.source_doc with sha256 diff"
```
### Task D3: Node-side cron — `lib/cron/sync_source_docs.js`
**Files:**
- Modify: `package.json` — add `node-cron`.
- Create: `lib/cron/index.js`
- Create: `lib/cron/sync_source_docs.js`
- Modify: `server.js` — start the cron in the CLI gate.
- Create: `tests/cron/sync_source_docs.test.js`
- [ ] **Step 1:** Add dep:
```bash
cd /project/src/void-v2 && npm i node-cron@^3
```
- [ ] **Step 2:** Failing test `tests/cron/sync_source_docs.test.js`:
```js
import { describe, it, expect, beforeEach, afterEach } from 'vitest';
import { resetDb } from '../helpers/db.js';
import { migrateUp } from '../../lib/db/migrate.js';
import { stopBoss } from '../helpers/boss.js';
import { pool } from '../../lib/db/pool.js';
import * as queue from '../../lib/jobs/queue.js';
import { registerWorkers } from '../../lib/jobs/index.js';
import { runSync } from '../../lib/cron/sync_source_docs.js';
import * as jobs from '../../lib/db/repos/jobs.js';
beforeEach(async () => {
await resetDb(); await migrateUp();
await queue.start(); await registerWorkers();
});
afterEach(async () => { await stopBoss(); });
describe('cron/sync_source_docs.runSync', () => {
it('enqueues sync.source_doc for each url-synced row', async () => {
const sp = (await pool.query(
`INSERT INTO spaces(slug, name) VALUES('s','S') RETURNING id`
)).rows[0].id;
const res = (await pool.query(
`INSERT INTO resources(space_id, slug, name, runtime_type) VALUES($1,'r','R','lxc') RETURNING id`,
[sp]
)).rows[0].id;
await pool.query(
`INSERT INTO source_docs(resource_id, name, upstream_url, sync_source) VALUES($1,'doc','https://example.com/r','url')`,
[res]
);
const enqueued = await runSync();
expect(enqueued).toBe(1);
const queued = await jobs.list({ name: 'sync.source_doc' });
expect(queued.length).toBe(1);
});
});
```
- [ ] **Step 3:** Run red.
- [ ] **Step 4:** Implement `lib/cron/sync_source_docs.js`:
```js
import { pool } from '../db/pool.js';
import * as queue from '../jobs/queue.js';
import { log } from '../log.js';
export async function runSync() {
const { rows } = await pool.query(
`SELECT id FROM source_docs WHERE sync_source = 'url'`
);
for (const r of rows) {
await queue.enqueue('sync.source_doc', { source_doc_id: r.id });
}
return rows.length;
}
```
- [ ] **Step 5:** Implement `lib/cron/index.js`:
```js
import cron from 'node-cron';
import { runSync } from './sync_source_docs.js';
import { log } from '../log.js';
export function startCron() {
// Daily at 03:00 local time
cron.schedule('0 3 * * *', async () => {
try {
const n = await runSync();
log.info({ enqueued: n }, 'cron sync.source_doc complete');
} catch (e) {
log.error({ err: e }, 'cron sync.source_doc failed');
}
});
log.info('cron started');
}
```
- [ ] **Step 6:** Wire into `server.js` — extend the CLI gate:
```js
// additions
import { startCron } from './lib/cron/index.js';
// Inside the `if (import.meta.url === ...)` block, after registerWorkers:
startCron();
```
- [ ] **Step 7:** Run green:
```bash
npx vitest run tests/cron/sync_source_docs.test.js
```
- [ ] **Step 8: Commit.**
```bash
git add package.json package-lock.json lib/cron/index.js lib/cron/sync_source_docs.js server.js tests/cron/sync_source_docs.test.js
git commit -m "feat(cron): daily sync.source_doc enqueue"
```
### Task D4: Version bump + CHANGELOG + completion doc
**Files:**
- Modify: `package.json` → `2.0.0-alpha.4`
- Modify: `server.js` → `VERSION = '2.0.0-alpha.4'`
- Modify: `tests/server.test.js` → assert `2.0.0-alpha.4`
- Modify: `CHANGELOG.md` — append `[2.0.0-alpha.4]` entry
- Create: `docs/plan-4-complete.md`
- [ ] **Step 1:** Bump version literals (three files). Use Edit.
- [ ] **Step 2:** Append CHANGELOG entry:
```markdown
## [2.0.0-alpha.4] — 2026-06-NN
### Added (Plan 4: Python void-workers)
- `void-workers.service` — Python 3.12 service alongside `void-server`
on CT 311. psycopg-based pg-boss client matches Node's claim/finish
semantics via `SELECT ... FOR UPDATE SKIP LOCKED`.
- **`extract.pdf`** — pdftotext first; per-page Tesseract OCR fallback
when extraction yields < 200 chars.
- **`extract.image`** — Tesseract OCR with English data.
- **`ingest.video`** — yt-dlp metadata + audio extraction + faster-whisper
transcription. CUDA at startup, CPU fallback on HA failover to Z3.
YouTube / youtu.be / Vimeo URLs route through `/api/capture` to
`ingest.video` instead of `ingest.url`.
- **`sync.source_doc`** — fetch upstream + sha256 diff + update
`body_text`. Node cron enqueues daily at 03:00.
- **Node `blob.js`** fans out to `extract.pdf` / `extract.image` after
creating PDF / image refs.
- **CT 311 infrastructure**: resized to 6 cores / 8 GB RAM, GPU device
nodes passed through (shared with CT 102's Ollama).
```
- [ ] **Step 3:** Create `docs/plan-4-complete.md` (use the Plan 3 doc as a template — sections: Date / Version / Tests / Commits / Snapshots / Scope per phase / Security findings handled / Open items / What's left after Plan 4).
- [ ] **Step 4:** Run full test suites green (Node + Python).
```bash
cd /project/src/void-v2 && npx vitest run
cd workers && DATABASE_URL=... pytest -v
```
- [ ] **Step 5: Commit.**
```bash
git add package.json server.js tests/server.test.js CHANGELOG.md docs/plan-4-complete.md
git commit -m "chore: version 2.0.0-alpha.4 + changelog + plan-4 completion doc"
```
### Task D5: Plan 4 close — snapshot + memory
- [ ] **Step 1:** Snapshot CT 310 + 311 — `plan4_complete_<timestamp>`.
- [ ] **Step 2:** Update memory `project_void_v2_execution.md` — mark Plan 4 complete; note alpha-4 source-built but not yet deployed (deploy is a separate user-OK decision per the standing rule).
---
## Spec coverage check
Every Plan 4 spec section maps to at least one task:
| Spec § | Task(s) |
|---|---|
| Phase A harness | A1, A2, A3, A4 |
| Phase B PDF + image | B1, B2, B3, B4, B5 |
| Phase C Whisper + yt-dlp + GPU | C1, C2, C3, C4, C5 |
| Phase D source-doc sync + version | D1, D2, D3, D4 |
| `safe_fetch` Python port | D1 |
| Concurrency env config | A1 (`config.py`) — applies across handlers |
| systemd `MemoryMax=6G` | A4 (`void-workers.service`) |
| CT 311 resize + GPU passthrough | C1 |
| YouTube routing in capture.js | C5 |
| `blob.js` fans out | B5 |
| Repo audit shim (`actor_kind='worker'`) | B2 (`repo.update_ref`), C4 (`repo.create_ref`), D2 (`repo.update_source_doc`) |
## Type & name consistency
- Worker name strings: `echo`, `extract.pdf`, `extract.image`, `ingest.video`, `sync.source_doc` — exact same strings used in `NAME` constants, registry keys, Node enqueue calls, and tests.
- Handler signature: `def handle(job_data: dict) -> dict` everywhere.
- Job payload shapes:
- `extract.pdf` / `extract.image`: `{ ref_id, blob_path }` (both producers Node-side use these keys; consumer reads same keys).
- `ingest.video`: `{ space_id, url }`.
- `sync.source_doc`: `{ source_doc_id }` (Node cron and Python worker both use this exact key).
- `repo.update_ref` / `repo.create_ref` / `repo.update_source_doc` / `repo.get_source_doc` — same names across all tasks that touch them.
Reference in New Issue View Git Blame Copy Permalink