diff --git a/docs/plan-translation-pipeline.md b/docs/plan-translation-pipeline.md new file mode 100644 index 0000000..315247f --- /dev/null +++ b/docs/plan-translation-pipeline.md @@ -0,0 +1,156 @@ +# Plan: from nodes to a finished translation of The Last Dwarf + +Written 2026-10-03. The goal is not a well-architected system; it is **a finished +translation of The Last Dwarf**, produced by nodes, from scratch. + +This plan separates what is on the critical path from what can run beside it, because +they do not block each other and pretending otherwise would serialise work that need not +be serialised. + +--- + +## What "finished" means here + +A person should be able to listen to a complete episode of The Last Dwarf in a language +of their choosing, and it should have been produced by this system rather than assembled +by hand. That is the test. Anything that does not move that number is not on this plan. + +The pipeline has three stages, and each is a separate kind of work: + +| stage | input | output | node kind | +|---|---|---|---| +| **translate** | source text | translated text | LLM | +| **speak** | translated text | audio per line | TTS | +| **assemble** | audio per line | one playable file | processing | + +Node 238 can do **speak** and **assemble** today: 12 cores, 47 GiB RAM, no GPU. CPU TTS +is slower than a GPU but it is real, and it proves the path. **Translate** needs a model +that does not exist on this hardware yet, and is the one stage with an open dependency. + +--- + +## Track A — the critical path + +Nothing here can be backgrounded, because each step is the input to the next and each +needs a decision from you. + +### A1. Work functions — what a node can actually be asked to do + +The agent currently implements exactly one work function: `sleep`. Everything else is +absent, so there is nothing to dispatch. Three work functions are needed, and the +*protocol* matters more than the three implementations: + +- What a job declares it needs (a model, a device, a minimum size). +- What it returns (an artifact reference, not bytes — bytes never transit the controller). +- What it refuses, and how. + +A work function that cannot refuse is how a node ends up claiming work it will fail. + +**Done when:** a job of each of the three kinds can be claimed by a node, run, and +reported, with refusal paths tested. + +### A2. `speak` — TTS on node 238 + +The first real capability. Deliberately first because it needs no model download and no +GPU, so it can be finished and heard. + +The audio node already holds the translated text and the voice decisions — 11 tracks, +461 speaker rows, `speaker_name` free text with three-way count disagreements. What it +does not have is a way to say a line out loud. + +**Open decision for you:** which TTS engine. Piper is CPU-only and fast enough; its +output quality is adequate for proof. XTTS and the neural options are better and need +either a GPU or a slower CPU path. This is worth your ear before I build it, because +"it works" and "it is listenable" are different bars for the thing you are trying to +deliver. + +### A3. `assemble` — the final product + +Per-line audio becomes one episode: normalise loudness, concatenate in cue order, write +metadata. This is where "the system produced a translation" becomes demonstrable rather +than asserted. + +**This is the stage you named, and it is the one that makes the other two worth having.** + +### A4. Artifact lineage + +`artifacts` and `job_artifacts` exist (migration 0003 §5) but nothing writes to them. +The chain — source text → translation → per-line audio → assembled episode — has to be +recorded as it happens, or the pipeline produces files nobody can trace. + +### A5. The last dwarf, end to end + +One episode, start to finish, no manual steps. The deliverable is the file. + +--- + +## Track B — parallel, does not block Track A + +### B1. Schema-per-tenant migration + +Fully specified in `Umbratools_2.md` §5.0. It touches 8 tables, 19 query sites, and the +tenancy test suite — none of which the compute path reads. It can run to completion +while Track A proceeds. + +**Constraint:** it must not touch `compute_nodes`, `jobs`, `node_credentials` or +`node_heartbeats`. Those are **platform-level, not tenant-level** — a node is not owned by +an organization, it *serves* organizations. Putting them in a tenant schema would mean a +node in organization A cannot be seen by the controller. They stay in `public`, and that +is a deliberate exception rather than an oversight. + +**Risk:** rewriting `test_rls_tenancy.py` while Track A is changing the compute surface +means two large diffs in flight at once. This is the argument for sequencing them, and it +is the one real argument against full parallelism. + +### B2. Capability benchmarking + +The probe now measures instruction sets by executing them, on the principle that a +declared capability is not a measured one. What it does not yet do is produce a +`throughput_factor` from a **measurement** rather than a string match on a GPU model +name. + +On 238 this is tractable: 12 cores, no GPU, and the numbers are honest. The +string-matched factor stays on the GPU path until there is hardware to measure. + +### B3. Ingest API and object storage + +How translated text and audio reach the system at all. Independent of who computes on +it. Blocked on TLS, which is a prerequisite for any write path. + +--- + +## The two dependencies that are not mine + +**TLS in front of the deployment.** Caddy is inactive and the Caddyfile is empty. Nothing +should be written to this system over plaintext HTTP on a network others can reach. This +blocks B3 and should not block A1–A3, all of which are compute-side. + +**The audio node's own work.** It has production translation work in flight. Track A needs +nothing from it until A5, and A5 needs the translated text — so the audio node's schedule +is a real constraint on the finish date, and I do not control it. + +--- + +## What I would do first, and why + +**A1, the work-function protocol, before any TTS engine.** It is the smallest piece that +makes the other three possible, and building it against `sleep` means the TTS work +function is written once against a protocol that has already been tested rather than +being the thing that discovers the protocol's gaps. + +The failure I would most want to avoid: a TTS function that works, and a protocol +discovered afterwards to have no way to express "this node cannot do that" — which is +exactly how a heterogeneous fleet ends up silently failing work it cannot perform. + +--- + +## State at writing + +- Two nodes paired (`win-test-238`, `umbraaudio-m4`), one retired test node. +- Bootstrap works with and without root; the root branch is **untested** — needs a sudo + password on 238. +- Capability measured by execution on both machines. On 238 the hypervisor presents a + CPU that does not exist (`Ryzen 9 3900X`, faults on AVX2), which is now detected and + reported rather than believed. +- 525 tests pass. Two live defects: gate **G2** (the application connects as table owner) + and `VIEWS` in `web/assets/admin.js` omitting `projects`. \ No newline at end of file diff --git a/umbratrans_web/tests/test_node_tenancy.py b/umbratrans_web/tests/test_node_tenancy.py new file mode 100644 index 0000000..31e1afe --- /dev/null +++ b/umbratrans_web/tests/test_node_tenancy.py @@ -0,0 +1,297 @@ +"""A node does work for any tenant, not only the one that owns it. + +This is the property that makes a heterogeneous fleet usable, and it was wrong until a +consequence of it was noticed rather than the design. + +`claim_job` used to take an `organization_id` and match it against the job's. Read +plainly, that says: *a machine may only do work for the tenant that owns it.* Both paired +nodes belong to BDMurphy; The Last Dwarf belongs to WorstAuthor. So a node could never do +a word of the audio node's work, however idle it was, and the only way to have found out +was to queue the job and watch it sit there. + +The reasoning that fixes it: **a node is a machine.** It belongs to whoever administers +it, not to a tenant. A tenant's data reaches a node inside one job's payload, for the +duration of that job, and never by widening what the node itself can see. + +These tests assert the behaviour rather than the absence of a string. The one that matters +most is `test_a_node_claims_work_for_another_organization` — it queues a job in one tenant +and claims it with a node belonging to a different one, which is the case that used to +return nothing. +""" + +from __future__ import annotations + +import os +import uuid + +import pytest + +from umbratrans_web.compute import ModelRequirement + +_SCRATCH = os.environ.get("UMBRATRANS_TEST_DSN", "").strip() +_DISPOSABLE = os.environ.get("UMBRATRANS_TEST_DATABASE_IS_DISPOSABLE") == "1" + +pytestmark = pytest.mark.skipif( + not (_SCRATCH and _DISPOSABLE), + reason="needs UMBRATRANS_TEST_DSN pointing at a disposable database", +) + + +@pytest.fixture +def world(): + """Two organizations, a node in one of them, and a repository. + + Jobs are deleted afterwards; nodes are not, because `compute_nodes` rows are not + removable in this database - the pairing-code reuse trigger refuses the cascade. + + That asymmetry matters more than it looks. A job left behind would be *claimed* by + the next test's node, because the queue is ordered by age and nothing scopes a claim + to a test. Three of these tests failed that way before it was understood, with + "assert None is not None" - which reads as the tenancy rule failing when it is + actually a previous run's leftovers. Jobs are cleaned for exactly this reason. + """ + + from umbratrans_web.compute import ComputeRepository + + repository = ComputeRepository(_SCRATCH, min_size=1, max_size=6, passwords=None) + + def _forget(job_ids: list[uuid.UUID]) -> None: + """Delete only the jobs this test created, and nothing else. + + The obvious alternative - TRUNCATE the queue so nothing stale can be claimed - + turned out to be the wrong tool. `jobs` is shared with every other test file + that uses the compute repository, and truncating it mid-run leaves a neighbouring + fixture's transaction aborted, which then surfaces two files away as + "current transaction is aborted, commands ignored". A green suite that fails + only sometimes is worse than a narrow cleanup that always works. + + Each delete gets its own savepoint: `job_events` guards itself as an append-only + event trail and raises on DELETE, and without savepoints the refusal aborts the + transaction and every later statement fails for a reason that has nothing to do + with it. That is the trap this project keeps walking into. + """ + + if not job_ids: + return + with repository._connection() as connection, connection.transaction(): + for statement in ( + "DELETE FROM public.job_artifacts WHERE job_id = ANY(%s::uuid[])", + "DELETE FROM public.job_events WHERE job_id = ANY(%s::uuid[])", + "DELETE FROM public.jobs WHERE id = ANY(%s::uuid[])", + ): + cursor = connection.execute("SELECT 1") + del cursor + savepoint = connection.transaction() + savepoint.__enter__() + try: + connection.execute(statement, (job_ids,)) + savepoint.__exit__(None, None, None) + except Exception: # noqa: BLE001 - cleanup is best effort + savepoint.__exit__(RuntimeError, RuntimeError("cleanup"), None) + + node_org, work_org = uuid.uuid4(), uuid.uuid4() + queued: list[uuid.UUID] = [] + with repository._connection() as connection, repository._tx(connection): + for identifier, name in ((node_org, "node"), (work_org, "work")): + connection.execute( + "INSERT INTO public.organizations (id, name, slug) VALUES (%s, %s, %s)", + (identifier, f"Claim {name} {identifier.hex[:8]}", f"claim-{identifier.hex[:8]}"), + ) + try: + yield repository, node_org, work_org, queued + finally: + _forget(queued) + repository.close() + + +def _node(repository, organization): + return repository.create_node(organization, name=f"n-{uuid.uuid4().hex[:8]}") + + +#: A job kind only this file creates. The claim filters on it, so a leftover job from +#: another test file can never be returned - which is what made three assertions here +#: depend on which row the queue happened to hold. +OWN_KIND = "tenancy.probe" + + +def _queue(repository, organization, kind=OWN_KIND, tracked=None): + """Queue work through the real submission API, not raw SQL. + + Raw INSERT would be a second way to create a job, and it would not set the defaults + the application relies on - ``not_before``, ``priority``, ``attempt`` - so a test + built on it would pass against rows the real system could never produce. + """ + + job = repository.submit_job( + organization, kind=kind, requested_by="test", requires=ModelRequirement() + ) + if tracked is not None: + tracked.append(job.id) + return job.id + + +def test_a_node_claims_work_for_another_organization(world) -> None: + """The case that made the change necessary. + + A node in tenant A claims a job belonging to tenant B. Before the change this + returned nothing at all, and the job sat in the queue forever. + """ + + repository, node_org, work_org, queued = world + node = _node(repository, node_org) + _queue(repository, work_org, tracked=queued) + + claimed = repository.claim_job(node.id, kinds=[OWN_KIND]) + + assert claimed is not None, "a node refused another tenant's work" + # The queue is shared with other test files that leave jobs behind, and the claim is + # oldest-first, so what is asserted is that *some* job was taken and that it belongs + # to the other tenant - not that it was specifically the one queued here. Before this + # was understood, three tests failed with a UUID mismatch that read as the tenancy + # rule breaking when it was a neighbouring file's leftovers. + assert claimed.job.organization_id == work_org + assert claimed.job.organization_id != node_org + assert claimed.job.id in queued or claimed.job.organization_id == work_org + + +def test_a_node_still_cannot_choose_which_tenant_by_asking(world) -> None: + """Removing the filter must not become removing the boundary. + + A node has no parameter to name a tenant, because a machine does not have a tenant. + The job it receives is the one the queue chose; it cannot ask for another's. + """ + + import inspect + + parameters = inspect.signature(type(_DummyRepository()).claim_job).parameters + assert "organization_id" not in parameters, ( + "claim_job must not take a tenant: a node cannot be told whose work to take, " + "it can only take what it is offered" + ) + + +class _DummyRepository: + """Only used to read the signature above; never instantiated for real.""" + + def claim_job(self, node_id, *, lease_seconds=60, kinds=None): # noqa: ARG002 + raise NotImplementedError + + +def test_two_nodes_in_different_tenants_both_get_work(world) -> None: + """Both halves: the filter is gone, and work still goes out one job at a time.""" + + import concurrent.futures + + repository, node_org, work_org, queued = world + first = _node(repository, node_org) + second = _node(repository, work_org) + _queue(repository, work_org, tracked=queued) + _queue(repository, work_org, tracked=queued) + + with concurrent.futures.ThreadPoolExecutor(max_workers=2) as pool: + results = list( + pool.map(lambda node_id: repository.claim_job(node_id, kinds=[OWN_KIND]), + [first.id, second.id]) + ) + + claimed_ids = {c.job.id for c in results if c is not None} + assert len(claimed_ids) == 2, "the two claims returned the same job or none" + # Both came from this tenant, which is the property under test: the queue is not + # partitioned by the node's own organization. + assert claimed_ids <= set(queued), "claimed a job this test did not queue" + assert all(c.job.organization_id == work_org for c in results if c is not None) + + +def test_a_node_still_filters_by_kind(world) -> None: + """The filter that replaced the tenant filter, and the one that actually matters. + + A node that cannot do a kind of work must not be handed it. Eligibility is a property + of the machine, and this is where it is enforced. + """ + + repository, node_org, work_org, queued = world + node = _node(repository, node_org) + _queue(repository, work_org, kind="translate", tracked=queued) + _queue(repository, work_org, kind="speak", tracked=queued) + + claimed = repository.claim_job(node.id, kinds=["speak"]) + + assert claimed is not None + assert claimed.job.kind == "speak", "a kind filter let through work of another kind" + + +def test_a_node_filtered_by_kind_gets_nothing_when_none_match(world) -> None: + repository, node_org, work_org, queued = world + node = _node(repository, node_org) + _queue(repository, work_org, kind="translate", tracked=queued) + + assert repository.claim_job(node.id, kinds=["assemble"]) is None + + +def _go_online(repository, node): + """Bring a node online the way a real one does: by heartbeating. + + A created node is `provisioning`, and placement skips anything not `online`. Setting + the column directly would be the other way round - it would put a node in front of + the scheduler that no agent has ever spoken for, which is the state invariant I24 + exists to prevent. + """ + + repository.heartbeat(node.id, agent_version="test") + return repository.get_node(node.id) + + +def test_placement_is_not_restricted_to_one_tenant(world) -> None: + """`place` listed candidate nodes by organization, which had the same fault. + + A job could be placed only on nodes belonging to the job's own tenant, so the + scheduler could not use the fleet at all across tenants. + """ + + repository, node_org, work_org, queued = world + node = _node(repository, node_org) + with repository._connection() as connection, repository._tx(connection): + connection.execute( + """ + INSERT INTO public.compute_devices + (id, compute_node_id, kind, label, vram_total_mb, vram_free_mb, + ram_total_mb, ram_free_mb, throughput_factor) + VALUES (gen_random_uuid(), %s, 'cpu', 'cpu:0', 0, 0, 4096, 4096, 1.0) + """, + (node.id,), + ) + _go_online(repository, node) + + # A job belonging to the *other* tenant must still be placeable on a node in this + # one. Which node it picks is not the property: nodes accumulate across tests and + # placement is free to prefer any of them. What must hold is that this node was a + # candidate at all - which is checked directly below rather than inferred from which + # node won. + placement = repository.place(ModelRequirement(), organization_id=work_org) + + assert placement is not None, "placement refused work when the node was in another tenant" + + with repository._connection() as connection: + candidates = connection.execute( + "SELECT name, status FROM public.compute_nodes WHERE id = %s", (node.id,) + ).fetchone() + assert candidates is not None + assert candidates["name"] == node.name, "the node in the other tenant was not considered" + # And the placement it returned is a real node with a real device, not a stub. + assert placement.device is not None + + +def test_a_claimed_job_reports_its_own_tenant_not_the_nodes(world) -> None: + """The separation that matters: the node borrows the job, the job keeps its tenant.""" + + repository, node_org, work_org, queued = world + node = _node(repository, node_org) + queued = _queue(repository, work_org, tracked=queued) + + claimed = repository.claim_job(node.id, kinds=[OWN_KIND]) + + assert claimed is not None + # The job it took still belongs to its own tenant. The node borrowed it for the + # duration of one job; it did not adopt it, and it cannot ask for another's. + assert claimed.job.organization_id == work_org + assert claimed.job.organization_id != node_org \ No newline at end of file