1917 lines
70 KiB
Python
1917 lines
70 KiB
Python
# Copyright 2026 Umbra Tools (umbratools.com)
|
|
# SPDX-License-Identifier: Apache-2.0
|
|
|
|
"""Compute routes, split by who is allowed to call them.
|
|
|
|
Two surfaces, because the credentials are not interchangeable and pretending
|
|
otherwise is how a node agent ends up holding a session token.
|
|
|
|
* ``/api/compute/*`` — the administrator's inventory and queue view. A session
|
|
with an administrative role, like every other ``uadmin`` route.
|
|
* ``/api/node/*`` — what a node agent may do. A **node credential** in the
|
|
``Authorization`` header, resolved to a node and an organization by the
|
|
credential itself.
|
|
|
|
The split is the whole point of ``node_credentials`` existing as something
|
|
narrower than an API key. A node credential can claim work and report a result.
|
|
It cannot read a project, list an organization, or reach anything in the
|
|
administrator surface, and there is no route under ``/api/node`` that would let it
|
|
try. §7.2, I28.
|
|
|
|
A node credential also **derives** its organization from the credential rather
|
|
than accepting one. That is what makes a forged tenant identifier unnecessary
|
|
instead of merely rejected, and it is why a node needs no database credential at
|
|
all - which matters more than it sounds once there are nodes outside the
|
|
building. §7.7, §13.15.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Any, Literal
|
|
from uuid import UUID
|
|
|
|
from fastapi import APIRouter, Depends, Header, Query, Request, Response
|
|
from pydantic import BaseModel, Field
|
|
|
|
from .app import require_admin
|
|
from .bootstrap import (
|
|
bootstrap_version,
|
|
render_bootstrap_template,
|
|
)
|
|
from .bootstrap import (
|
|
render as render_bootstrap,
|
|
)
|
|
from .compute import (
|
|
DEFAULT_LEASE_SECONDS,
|
|
NODE_SCOPES,
|
|
ComputeNode,
|
|
ModelRequirement,
|
|
)
|
|
from .errors import (
|
|
AuthenticationRequiredError,
|
|
ForbiddenError,
|
|
InputError,
|
|
NotFoundError,
|
|
PairingError,
|
|
)
|
|
from .models import Principal
|
|
from .schemas import RequestModel
|
|
|
|
#: The administrator's inventory and queue view. A session with an administrative
|
|
#: role, like every other ``uadmin`` route.
|
|
admin_router = APIRouter()
|
|
|
|
#: What a node agent may do, and nothing else. A node credential.
|
|
#:
|
|
#: These are two routers rather than one router mounted twice, and the difference
|
|
#: is a security boundary. Mounting one router at two prefixes publishes every
|
|
#: route at both, which would put node creation and credential issuance behind
|
|
#: node-credential authentication - letting a node hand itself more authority.
|
|
#: The route lists below are asserted disjoint in ``test_compute_routes.py`` so a
|
|
#: future route cannot be added to one and reach the other by accident.
|
|
agent_router = APIRouter()
|
|
|
|
MAX_PAGE_SIZE = 200
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# Schemas
|
|
# --------------------------------------------------------------------------
|
|
|
|
|
|
class DeviceReport(BaseModel):
|
|
"""What a node measured about one of its devices.
|
|
|
|
``throughput_factor`` is measured by the node, not declared by a human.
|
|
Declared factors are guesses that go stale the moment quantisation or
|
|
context length changes, and a stale factor sends work to the wrong machine.
|
|
"""
|
|
|
|
kind: str = Field(pattern="^(cpu|gpu_discrete|gpu_integrated|npu|tpu)$")
|
|
label: str = Field(min_length=1, max_length=64)
|
|
vram_total_mb: int = Field(default=0, ge=0)
|
|
vram_free_mb: int = Field(default=0, ge=0)
|
|
ram_total_mb: int = Field(default=0, ge=0)
|
|
ram_free_mb: int = Field(default=0, ge=0)
|
|
compute_units: float | None = None
|
|
throughput_factor: float = Field(default=1.0, gt=0)
|
|
supports_accelerators: list[str] = Field(default_factory=list)
|
|
|
|
|
|
class DeviceResponse(BaseModel):
|
|
id: UUID
|
|
kind: str
|
|
label: str
|
|
vram_total_mb: int
|
|
vram_free_mb: int
|
|
ram_total_mb: int
|
|
ram_free_mb: int
|
|
compute_units: float | None = None
|
|
throughput_factor: float
|
|
supports_accelerators: list[str]
|
|
busy: bool
|
|
|
|
|
|
class NodeResponse(BaseModel):
|
|
"""A node, and the two facts a person needs to know if it is usable.
|
|
|
|
``schedulable`` already accounts for a device being out of memory, so the
|
|
console does not have to reimplement the rule. And ``status_reason`` carries
|
|
the explanation, because "not schedulable" with no reason is the state that
|
|
generates a support question.
|
|
"""
|
|
|
|
id: UUID
|
|
name: str
|
|
is_controller: bool
|
|
trust_domain: str
|
|
status: str
|
|
locality: str | None = None
|
|
transport: str = "https"
|
|
agent_version: str | None = None
|
|
last_heartbeat_at: str | None = None
|
|
schedulable: bool
|
|
status_reason: str
|
|
#: What the database column says, when it differs from `status`. `status` is
|
|
#: derived from the stored value and the clock, so a node that has gone quiet
|
|
#: reads `offline` while the row still says `online`. Without this field an
|
|
#: operator comparing the API to the database has no way to tell a stale cache
|
|
#: from a contradiction, which is the ambiguity worth spending a field on.
|
|
stored_status: str | None = None
|
|
devices: list[DeviceResponse] = Field(default_factory=list)
|
|
|
|
|
|
class CreateNodeRequest(RequestModel):
|
|
name: str = Field(min_length=1, max_length=120)
|
|
is_controller: bool = False
|
|
trust_domain: str = Field(default="local", pattern="^(local|remote|cloud)$")
|
|
locality: str | None = Field(default=None, max_length=120)
|
|
transport: str = Field(default="https", pattern="^(https|ssh|vpn)$")
|
|
|
|
|
|
class BootstrapResponse(BaseModel):
|
|
"""A rendered bootstrap, and the single-use code inside it.
|
|
|
|
``code`` is here once and never again: the controller stores only an Argon2id hash,
|
|
so this response is the only moment the plaintext exists outside the shell script
|
|
that embeds it.
|
|
"""
|
|
|
|
node_name: str
|
|
code: str
|
|
code_hint: str
|
|
expires_at: datetime
|
|
script: str
|
|
command: str
|
|
|
|
|
|
class BootstrapRequest(RequestModel):
|
|
"""What an administrator needs to say to render a bootstrap.
|
|
|
|
``as_root`` selects the system install over the user install. It does not grant
|
|
anything: the script checks for root when it runs and takes the branch that is
|
|
actually available, so a false value on a machine where the operator happens to be
|
|
root still produces a working system install rather than a confusing mismatch.
|
|
"""
|
|
|
|
node_name: str = Field(min_length=1, max_length=120)
|
|
locality: str | None = Field(default=None, max_length=120)
|
|
trust_domain: str = Field(default="local", pattern="^(local|remote|cloud)$")
|
|
ttl_seconds: int = Field(default=1800, ge=60, le=1800)
|
|
controller_url: str = Field(default="http://192.168.123.14:8442", max_length=300)
|
|
as_root: bool = False
|
|
|
|
|
|
class PairingCodeRequest(RequestModel):
|
|
"""An administrator naming the machine they are standing at.
|
|
|
|
``ttl_seconds`` defaults to ten minutes and is bounded below at 60s. A code is
|
|
short-lived on purpose: it is a one-time bearer secret that a human reads aloud or
|
|
types by hand, so it should stop working the moment it stops being convenient.
|
|
"""
|
|
|
|
node_name: str = Field(min_length=1, max_length=120)
|
|
locality: str | None = Field(default=None, max_length=120)
|
|
trust_domain: str = Field(default="local", pattern="^(local|remote|cloud)$")
|
|
ttl_seconds: int = Field(default=600, ge=60, le=1800)
|
|
|
|
|
|
class PairingCodeResponse(BaseModel):
|
|
"""The only time the code exists. Hash at rest, so it cannot be re-read."""
|
|
|
|
code: str
|
|
code_hint: str
|
|
node_name: str
|
|
expires_at: datetime
|
|
|
|
|
|
class NodePairRequest(RequestModel):
|
|
"""What a node sends to pair. It has no credential yet - that is the point."""
|
|
|
|
code: str = Field(min_length=1, max_length=16)
|
|
node_name: str = Field(min_length=1, max_length=120)
|
|
agent_version: str | None = Field(default=None, max_length=64)
|
|
locality: str | None = Field(default=None, max_length=120)
|
|
|
|
mac: str | None = Field(
|
|
default=None,
|
|
max_length=32,
|
|
pattern=r"^[0-9A-Fa-f]{2}(:[0-9A-Fa-f]{2}){5}$",
|
|
)
|
|
"""The node's MAC, which is what makes its name unique.
|
|
|
|
Optional, and deliberately so: a machine that cannot report one still pairs, and
|
|
falls back to the letter-suffix sequence on its own. What it must not do is claim a
|
|
name that already belongs to another machine, so the fallback is a collision
|
|
resolution rather than a silent overwrite - see :func:`resolve_node_name`.
|
|
"""
|
|
|
|
|
|
class NodePairResponse(BaseModel):
|
|
"""A paired node and its credential. The key is shown once and never stored."""
|
|
|
|
node_id: UUID
|
|
node_name: str
|
|
base_name: str | None = None
|
|
key_prefix: str
|
|
key: str
|
|
scopes: list[str]
|
|
|
|
|
|
class NodeCredentialRequest(RequestModel):
|
|
agent_name: str = Field(min_length=1, max_length=64)
|
|
scopes: list[str] = Field(min_length=1)
|
|
|
|
|
|
class NodeCredentialResponse(BaseModel):
|
|
"""The only time the key exists. It is not stored and cannot be re-read."""
|
|
|
|
node_id: UUID
|
|
key_prefix: str
|
|
key: str
|
|
scopes: list[str]
|
|
scopes_available: list[str]
|
|
note: str
|
|
|
|
|
|
class SetNodeStatusRequest(RequestModel):
|
|
status: str = Field(pattern="^(provisioning|draining|offline|retired)$")
|
|
"""The states an administrator may put a node into by hand.
|
|
|
|
``removed`` is deliberately not here: it is what :func:`remove_node` sets, and it is
|
|
not a state a node can be argued back out of. Allowing it in this field would let
|
|
"remove this machine" be typed as a status change, which would skip the credential
|
|
revocation and the release of its work - the two things that make removal mean
|
|
anything.
|
|
"""
|
|
|
|
|
|
class ModelRequest(RequestModel):
|
|
slug: str = Field(min_length=1, max_length=120)
|
|
kind: str = Field(default="llm", pattern="^(llm|tts|asr|embedding|vision|utility)$")
|
|
vram_min_mb: int = Field(default=0, ge=0)
|
|
vram_preferred_mb: int = Field(default=0, ge=0)
|
|
ram_min_mb: int = Field(default=0, ge=0)
|
|
requires_accelerators: list[str] = Field(default_factory=list)
|
|
reference_throughput: float = Field(default=1.0, gt=0)
|
|
load_seconds: int = Field(default=0, ge=0)
|
|
|
|
|
|
class ModelResponse(BaseModel):
|
|
id: UUID
|
|
slug: str
|
|
kind: str
|
|
vram_min_mb: int
|
|
vram_preferred_mb: int
|
|
ram_min_mb: int
|
|
requires_accelerators: list[str]
|
|
reference_throughput: float
|
|
load_seconds: int
|
|
|
|
|
|
class JobRequirement(RequestModel):
|
|
"""What a job needs.
|
|
|
|
``vram_min_mb`` and ``ram_min_mb`` are eligibility floors, not wishes. A job
|
|
that names a floor nothing satisfies is ``blocked`` with a reason rather than
|
|
left in a queue looking like it is progressing. §7.3.1
|
|
"""
|
|
|
|
vram_min_mb: int = Field(default=0, ge=0)
|
|
ram_min_mb: int = Field(default=0, ge=0)
|
|
models: list[str] = Field(default_factory=list)
|
|
trust_domains: list[str] = Field(default_factory=list)
|
|
accelerators: list[str] = Field(default_factory=list)
|
|
max_runtime_s: int | None = None
|
|
|
|
|
|
class SubmitJobRequest(RequestModel):
|
|
kind: str = Field(min_length=1, max_length=64)
|
|
requires: JobRequirement = Field(default_factory=JobRequirement)
|
|
payload: dict[str, Any] = Field(default_factory=dict)
|
|
priority: int = Field(default=100, ge=0, le=1000)
|
|
project_id: UUID | None = None
|
|
lang: str | None = Field(default=None, max_length=32)
|
|
idempotency_key: str | None = Field(default=None, max_length=200)
|
|
max_attempts: int = Field(default=3, ge=1, le=10)
|
|
#: When true, the job is refused as soon as nothing can run it, instead of
|
|
#: waiting for a node that may never appear.
|
|
fail_if_unplaceable: bool = True
|
|
|
|
|
|
class JobResponse(BaseModel):
|
|
id: UUID
|
|
kind: str
|
|
status: str
|
|
requested_by: str
|
|
priority: int
|
|
requires: JobRequirement
|
|
payload: dict[str, Any]
|
|
project_id: UUID | None = None
|
|
lang: str | None = None
|
|
lease_owner: UUID | None = None
|
|
attempt: int
|
|
max_attempts: int
|
|
parent_job_id: UUID | None = None
|
|
idempotency_key: str | None = None
|
|
result: dict[str, Any] | None = None
|
|
error: str | None = None
|
|
created_at: str | None = None
|
|
finished_at: str | None = None
|
|
|
|
|
|
class PlacementResponse(BaseModel):
|
|
node_id: str
|
|
node_name: str
|
|
device_id: str
|
|
device_kind: str
|
|
reason: str
|
|
resident: bool
|
|
would_evict: list[str]
|
|
estimated_seconds: int | None = None
|
|
|
|
|
|
class ClaimedJobResponse(BaseModel):
|
|
"""The job, the plan, and how long the lease is good for.
|
|
|
|
The plan travels with the job so the node knows whether it is being asked to
|
|
load a model or evict one *before* it accepts. A node that discovers the
|
|
eviction after starting has already wasted the time.
|
|
"""
|
|
|
|
job: JobResponse
|
|
placement: PlacementResponse
|
|
lease_expires_at: str
|
|
renew_after_seconds: int
|
|
|
|
|
|
class HeartbeatRequest(BaseModel):
|
|
agent_version: str | None = Field(default=None, max_length=64)
|
|
devices: list[DeviceReport] = Field(default_factory=list)
|
|
resident_models: list[str] = Field(default_factory=list)
|
|
active_jobs: int = Field(default=0, ge=0)
|
|
measurements: dict[str, Any] = Field(default_factory=dict)
|
|
|
|
|
|
class HeartbeatResponse(BaseModel):
|
|
"""The node as the controller now sees it.
|
|
|
|
A node is promoted to ``online`` here and nowhere else, which is what makes
|
|
``online`` mean "this node has spoken" rather than "this row exists". (I24)
|
|
"""
|
|
|
|
node_id: UUID
|
|
name: str
|
|
status: str
|
|
schedulable: bool
|
|
status_reason: str
|
|
devices: list[DeviceResponse]
|
|
heartbeat_interval_seconds: int = 10
|
|
|
|
|
|
class CompleteJobRequest(BaseModel):
|
|
result: dict[str, Any] = Field(default_factory=dict)
|
|
|
|
|
|
class FailJobRequest(BaseModel):
|
|
error: str = Field(min_length=1, max_length=2000)
|
|
retry: bool = True
|
|
|
|
|
|
class ChildJobRequest(BaseModel):
|
|
kind: str = Field(min_length=1, max_length=64)
|
|
requires: JobRequirement = Field(default_factory=JobRequirement)
|
|
payload: dict[str, Any] = Field(default_factory=dict)
|
|
priority: int = Field(default=100, ge=0, le=1000)
|
|
idempotency_key: str | None = Field(default=None, max_length=200)
|
|
|
|
|
|
class DecomposeRequest(BaseModel):
|
|
children: list[ChildJobRequest] = Field(min_length=1, max_length=200)
|
|
|
|
|
|
class RenewResponse(BaseModel):
|
|
job_id: UUID
|
|
lease_expires_at: str | None
|
|
ok: bool
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# Node-credential authentication
|
|
# --------------------------------------------------------------------------
|
|
|
|
|
|
class NodeContext:
|
|
"""Who the node is, resolved from the credential and nothing else."""
|
|
|
|
__slots__ = ("node_id", "organization_id", "scopes")
|
|
|
|
def __init__(self, node_id: UUID, organization_id: UUID, scopes: list[str]) -> None:
|
|
self.node_id = node_id
|
|
self.organization_id = organization_id
|
|
self.scopes = scopes
|
|
|
|
def require(self, scope: str) -> None:
|
|
if scope not in self.scopes:
|
|
# 403 with the scope named, because the agent is a program and can
|
|
# read this; unlike a human caller, it has no ambiguity to protect
|
|
# against by staying vague.
|
|
raise ForbiddenError(
|
|
"node_scope_required",
|
|
f"This node credential does not carry '{scope}'.",
|
|
)
|
|
|
|
|
|
def compute_repository(request: Request):
|
|
"""The compute repository, built once per application.
|
|
|
|
Separate from the account repository on purpose: the account boundary refuses
|
|
anything but the PostgreSQL adapter, and this is a second adapter for a
|
|
second concern rather than a way around the first.
|
|
"""
|
|
|
|
cached = getattr(request.app.state, "compute_repository", None)
|
|
if cached is not None:
|
|
return cached
|
|
from .compute import build_compute_repository
|
|
|
|
repository = build_compute_repository(
|
|
request.app.state.settings,
|
|
getattr(request.app.state, "password_service", None),
|
|
)
|
|
request.app.state.compute_repository = repository
|
|
return repository
|
|
|
|
|
|
def audio_import_store(request: Request):
|
|
"""The PostgreSQL audio-import store, built once per application.
|
|
|
|
Only a commit needs it, so it is built on the commit path rather than at
|
|
startup: a dry run is pure validation and must succeed on a deployment whose
|
|
database is not reachable. There is deliberately no in-memory fallback here,
|
|
because a commit that silently names nothing is the exact failure this
|
|
adapter exists to prevent.
|
|
"""
|
|
|
|
cached = getattr(request.app.state, "audio_import_store", None)
|
|
if cached is not None:
|
|
return cached
|
|
from .audio_import_postgres import PostgresAudioImportStore
|
|
|
|
store = PostgresAudioImportStore.from_settings(request.app.state.settings)
|
|
request.app.state.audio_import_store = store
|
|
return store
|
|
|
|
|
|
def require_node(
|
|
request: Request,
|
|
authorization: str | None = Header(default=None),
|
|
) -> NodeContext:
|
|
"""Resolve a node credential.
|
|
|
|
The organization comes from the node the credential points at. A request
|
|
cannot name one, so a forged tenant identifier has nothing to attach to.
|
|
"""
|
|
|
|
if not authorization or not authorization.lower().startswith("bearer "):
|
|
raise AuthenticationRequiredError(
|
|
"node_credential_required",
|
|
"A node credential is required: Authorization: Bearer <key>.",
|
|
)
|
|
key = authorization.split(" ", 1)[1].strip()
|
|
repository = compute_repository(request)
|
|
resolved = repository.resolve_node_credential(key)
|
|
if resolved is None:
|
|
raise AuthenticationRequiredError(
|
|
"node_credential_invalid",
|
|
"The node credential is not valid, or has been revoked.",
|
|
)
|
|
node_id, organization_id, scopes = resolved
|
|
repository.mark_credential_used(node_id)
|
|
return NodeContext(node_id, organization_id, scopes)
|
|
|
|
|
|
def _device_dict(device) -> DeviceResponse:
|
|
return DeviceResponse(
|
|
id=device.id,
|
|
kind=device.kind,
|
|
label=device.label,
|
|
vram_total_mb=device.vram_total_mb,
|
|
vram_free_mb=device.vram_free_mb,
|
|
ram_total_mb=device.ram_total_mb,
|
|
ram_free_mb=device.ram_free_mb,
|
|
compute_units=device.compute_units,
|
|
throughput_factor=device.throughput_factor,
|
|
supports_accelerators=list(device.supports_accelerators),
|
|
busy=device.is_busy,
|
|
)
|
|
|
|
|
|
def _node_reason(node: ComputeNode) -> str:
|
|
"""Why a node is or is not schedulable, in one sentence.
|
|
|
|
The console shows this verbatim. A node that is ``online`` with every device
|
|
out of memory needs to say so, because "online" and "can be given work" are
|
|
different facts and only the first is a status.
|
|
"""
|
|
|
|
if node.status == "provisioning":
|
|
return "provisioning: the agent has never authenticated"
|
|
if node.status == "retired":
|
|
return "retired: the node has been removed from service"
|
|
if node.status == "removed":
|
|
return (
|
|
"removed: the machine is gone. Its credentials are revoked and it cannot be "
|
|
"given work; this row is kept only because the job history refers to it."
|
|
)
|
|
if node.status == "draining":
|
|
return "draining: an administrator stopped new claims; running work finishes"
|
|
if node.status == "offline":
|
|
if node.stored_status == "online":
|
|
# Derived, not stored: the column still says online and the node has
|
|
# gone quiet. Saying "offline: no recent heartbeat" alone would read as
|
|
# though somebody had marked it offline, and an operator would go looking
|
|
# for that decision instead of for the dead agent.
|
|
return "offline: no recent heartbeat (derived; the stored status is still 'online')"
|
|
return "offline: no recent heartbeat"
|
|
if node.status != "online":
|
|
return f"status {node.status}"
|
|
if not node.devices:
|
|
return "online: no devices reported yet"
|
|
if all(device.is_busy for device in node.devices):
|
|
held = ", ".join(
|
|
f"{device.label} ({device.vram_free_mb} MiB free of {device.vram_total_mb})"
|
|
for device in node.devices
|
|
)
|
|
return f"online: out of memory on every device - {held}"
|
|
return "online: accepting work"
|
|
|
|
|
|
def _node_response(node: ComputeNode) -> NodeResponse:
|
|
return NodeResponse(
|
|
id=node.id,
|
|
name=node.name,
|
|
is_controller=node.is_controller,
|
|
trust_domain=node.trust_domain,
|
|
status=node.status,
|
|
locality=node.locality,
|
|
transport=node.transport,
|
|
agent_version=node.agent_version,
|
|
last_heartbeat_at=node.last_heartbeat_at.isoformat() if node.last_heartbeat_at else None,
|
|
schedulable=node.is_schedulable,
|
|
# Only reported when it differs, so an ordinary online node's response is
|
|
# unchanged and a stale one says plainly that the column disagrees.
|
|
stored_status=node.stored_status if node.stored_status != node.status else None,
|
|
status_reason=_node_reason(node),
|
|
devices=[_device_dict(device) for device in node.devices],
|
|
)
|
|
|
|
|
|
def _job_response(job) -> JobResponse:
|
|
return JobResponse(
|
|
id=job.id,
|
|
kind=job.kind,
|
|
status=job.status,
|
|
requested_by=job.requested_by,
|
|
priority=job.priority,
|
|
requires=JobRequirement(**job.requires.to_json()),
|
|
payload=dict(job.payload or {}),
|
|
project_id=job.project_id,
|
|
lang=job.lang,
|
|
lease_owner=job.lease_owner,
|
|
attempt=job.attempt,
|
|
max_attempts=job.max_attempts,
|
|
parent_job_id=job.parent_job_id,
|
|
idempotency_key=job.idempotency_key,
|
|
result=dict(job.result) if job.result else None,
|
|
error=job.error,
|
|
created_at=job.created_at.isoformat() if job.created_at else None,
|
|
finished_at=job.finished_at.isoformat() if job.finished_at else None,
|
|
)
|
|
|
|
|
|
def _organization_of(principal: Principal) -> UUID:
|
|
"""The organization an administrator's request acts within.
|
|
|
|
A platform administrator with no organization of their own acts within the one
|
|
they name, which is why the node routes take it as a query parameter rather
|
|
than deriving it. Everywhere else this is pinned.
|
|
"""
|
|
|
|
if principal.organization_id is not None:
|
|
return principal.organization_id
|
|
raise InputError(
|
|
"organization_required",
|
|
"A platform administrator must name the organization: ?organization_id=…",
|
|
)
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# Administrator surface: /api/compute
|
|
# --------------------------------------------------------------------------
|
|
|
|
|
|
@admin_router.get("/nodes", response_model=list[NodeResponse])
|
|
def list_nodes(
|
|
request: Request,
|
|
principal: Principal = Depends(require_admin),
|
|
organization_id: UUID | None = Query(default=None),
|
|
) -> list[NodeResponse]:
|
|
"""The inventory. The console's compute page is a view of this list."""
|
|
|
|
repository = compute_repository(request)
|
|
target = organization_id or _organization_of(principal)
|
|
return [_node_response(node) for node in repository.list_nodes(target)]
|
|
|
|
|
|
@admin_router.post("/nodes", response_model=NodeResponse, status_code=201)
|
|
def create_node(
|
|
payload: CreateNodeRequest,
|
|
request: Request,
|
|
principal: Principal = Depends(require_admin),
|
|
organization_id: UUID | None = Query(default=None),
|
|
) -> NodeResponse:
|
|
"""Add a machine. It is ``provisioning`` until its agent heartbeats. (I24)
|
|
|
|
Creating a node and having it appear ready would be the more convenient
|
|
behaviour and the wrong one: a node whose agent cannot reach the controller
|
|
would look available and silently swallow every job dispatched to it.
|
|
"""
|
|
|
|
repository = compute_repository(request)
|
|
target = organization_id or _organization_of(principal)
|
|
node = repository.create_node(
|
|
target,
|
|
name=payload.name,
|
|
is_controller=payload.is_controller,
|
|
trust_domain=payload.trust_domain,
|
|
locality=payload.locality,
|
|
transport=payload.transport,
|
|
)
|
|
return _node_response(node)
|
|
|
|
|
|
@admin_router.post("/pairing-codes", response_model=PairingCodeResponse, status_code=201)
|
|
def create_pairing_code(
|
|
payload: PairingCodeRequest,
|
|
request: Request,
|
|
principal: Principal = Depends(require_admin),
|
|
organization_id: UUID | None = Query(default=None),
|
|
) -> PairingCodeResponse:
|
|
"""Mint a single-use code for a named node, for that node to exchange.
|
|
|
|
The alternative - reading a credential out of a file and typing it into a terminal
|
|
on another machine - put a live credential into shell history and required the
|
|
controller to be a courier. This does not: the code is short-lived, single-use,
|
|
bound to the node name, and worthless once spent.
|
|
|
|
The response is the only place the plaintext code exists. It is never logged and
|
|
never stored; the row keeps an Argon2id hash.
|
|
"""
|
|
|
|
repository = compute_repository(request)
|
|
minted = repository.create_pairing_code(
|
|
organization_id or _organization_of(principal),
|
|
payload.node_name,
|
|
locality=payload.locality,
|
|
trust_domain=payload.trust_domain,
|
|
ttl_seconds=payload.ttl_seconds,
|
|
)
|
|
return PairingCodeResponse(**minted)
|
|
|
|
|
|
@admin_router.post("/nodes/{node_id}/credentials", response_model=NodeCredentialResponse, status_code=201)
|
|
def issue_credential(
|
|
node_id: UUID,
|
|
payload: NodeCredentialRequest,
|
|
request: Request,
|
|
principal: Principal = Depends(require_admin),
|
|
) -> NodeCredentialResponse:
|
|
"""Issue a node credential. The key is shown once and never stored.
|
|
|
|
Scopes are restricted to the node set. A scope that could read project
|
|
content would make this a session token with a different name, and the whole
|
|
reason ``node_credentials`` is a separate table is that it is not one.
|
|
"""
|
|
|
|
repository = compute_repository(request)
|
|
prefix, key = repository.issue_node_credential(
|
|
node_id, agent_name=payload.agent_name, scopes=payload.scopes
|
|
)
|
|
return NodeCredentialResponse(
|
|
node_id=node_id,
|
|
key_prefix=prefix,
|
|
key=key,
|
|
scopes=sorted(set(payload.scopes)),
|
|
scopes_available=sorted(NODE_SCOPES),
|
|
note="Shown once. It is stored only as an argon2id hash and cannot be re-read.",
|
|
)
|
|
|
|
|
|
@admin_router.patch("/nodes/{node_id}/status", response_model=NodeResponse)
|
|
def set_node_status(
|
|
node_id: UUID,
|
|
payload: SetNodeStatusRequest,
|
|
request: Request,
|
|
principal: Principal = Depends(require_admin),
|
|
) -> NodeResponse:
|
|
"""Drain, retire, or return a node to service.
|
|
|
|
``draining`` is the tool that makes a reboot controlled: it stops new claims
|
|
and lets running work finish, rather than producing a batch of expired leases.
|
|
"""
|
|
|
|
repository = compute_repository(request)
|
|
return _node_response(repository.set_node_status(node_id, payload.status))
|
|
|
|
|
|
class ReasonRequest(RequestModel):
|
|
"""Why, and who. Both recorded.
|
|
|
|
A cancellation with no recorded cause is the one thing that makes "why did this job
|
|
stop" unanswerable a month later, when the difference between a withdrawn machine and
|
|
a superseded one is exactly what an operator is trying to establish.
|
|
"""
|
|
|
|
reason: str = Field(min_length=1, max_length=500)
|
|
actor: str | None = Field(default=None, max_length=120)
|
|
|
|
|
|
class RemoveNodeRequest(ReasonRequest):
|
|
"""Removal takes the same fields; the class exists so the two read differently.
|
|
|
|
Both are required rather than defaulted for the same reason: dead hardware and a
|
|
decommissioned licence look identical six months afterwards.
|
|
"""
|
|
|
|
|
|
class AssignmentResponse(BaseModel):
|
|
"""One machine's attempt at one job, as the controller sees it.
|
|
|
|
``artifacts_verified_at`` is the field that matters and it is deliberately not
|
|
derivable from ``state``: a ``succeeded`` assignment whose timestamp is null is a node
|
|
that says it finished and has not been confirmed, and the two must not look alike.
|
|
"""
|
|
|
|
id: UUID
|
|
job_id: UUID
|
|
node_id: UUID
|
|
state: str
|
|
attempt: int
|
|
assigned_at: datetime | None = None
|
|
started_at: datetime | None = None
|
|
finished_at: datetime | None = None
|
|
lease_expires_at: datetime | None = None
|
|
artifacts_verified_at: datetime | None = None
|
|
verified: bool
|
|
error: str | None = None
|
|
|
|
|
|
class AssignmentOutcomeResponse(BaseModel):
|
|
"""The result of an operation on an assignment, including what verification wanted."""
|
|
|
|
assignment: AssignmentResponse
|
|
verified: bool
|
|
job_status: str
|
|
missing: list[str] = Field(default_factory=list)
|
|
artifacts_recorded: list[str] = Field(default_factory=list)
|
|
superseded_nodes: list[str] = Field(default_factory=list)
|
|
|
|
|
|
def _assignment_response(assignment: Any) -> AssignmentResponse:
|
|
return AssignmentResponse(
|
|
id=assignment.id,
|
|
job_id=assignment.job_id,
|
|
node_id=assignment.node_id,
|
|
state=assignment.state,
|
|
attempt=assignment.attempt,
|
|
assigned_at=assignment.assigned_at,
|
|
started_at=assignment.started_at,
|
|
finished_at=assignment.finished_at,
|
|
lease_expires_at=assignment.lease_expires_at,
|
|
artifacts_verified_at=assignment.artifacts_verified_at,
|
|
verified=assignment.is_verified,
|
|
error=assignment.error,
|
|
)
|
|
|
|
|
|
def _outcome_response(outcome: Any) -> AssignmentOutcomeResponse:
|
|
return AssignmentOutcomeResponse(
|
|
assignment=_assignment_response(outcome.assignment),
|
|
verified=outcome.verified,
|
|
job_status=outcome.job_status,
|
|
missing=list(outcome.missing),
|
|
artifacts_recorded=list(outcome.artifacts_recorded),
|
|
superseded_nodes=list(outcome.superseded_nodes),
|
|
)
|
|
|
|
|
|
@admin_router.post("/nodes/{node_id}/revoke", response_model=NodeResponse)
|
|
def revoke_node_credentials(
|
|
node_id: UUID,
|
|
payload: RemoveNodeRequest,
|
|
request: Request,
|
|
principal: Principal = Depends(require_admin),
|
|
) -> NodeResponse:
|
|
"""Cut a node off without removing it. The machine stays, the credential dies.
|
|
|
|
This is the right tool for a key that may be compromised, or a machine that is going
|
|
to be reimaged: the node stops being able to authenticate immediately, and nothing
|
|
about its history is lost. `POST /nodes/{node_id}/remove` is the one that also takes
|
|
the machine out of the fleet.
|
|
"""
|
|
|
|
repository = compute_repository(request)
|
|
repository.revoke_node_credentials(
|
|
node_id, reason=payload.reason, actor=payload.actor
|
|
)
|
|
node = repository.get_node(node_id)
|
|
if node is None:
|
|
raise NotFoundError("compute_node_not_found")
|
|
return _node_response(node)
|
|
|
|
|
|
@admin_router.post("/nodes/{node_id}/remove", response_model=NodeResponse)
|
|
def remove_node(
|
|
node_id: UUID,
|
|
payload: RemoveNodeRequest,
|
|
request: Request,
|
|
principal: Principal = Depends(require_admin),
|
|
) -> NodeResponse:
|
|
"""Take a machine out of the fleet for good, for hardware that is not coming back.
|
|
|
|
Revokes its credentials, deletes its devices, heartbeats and model reports, and
|
|
releases any work it was holding so that work is reassignable rather than stalled
|
|
behind a machine nobody is going to switch on.
|
|
|
|
The row itself stays, and cannot be deleted: `job_events.node_id` refers to it, and
|
|
the event trail refuses every update, so removing the row would mean either losing
|
|
the record of what the machine did or weakening the guarantee that the record is
|
|
immutable. The status becomes ``removed``, which is terminal - there is no
|
|
administrative path back out of it, because a machine that has been written off is
|
|
not coming back and a node that can be un-removed is a node nobody can be sure of.
|
|
"""
|
|
|
|
repository = compute_repository(request)
|
|
return _node_response(repository.remove_node(node_id, reason=payload.reason, actor=payload.actor))
|
|
|
|
|
|
class AssignJobRequest(RequestModel):
|
|
"""Who should get this job, and how many of them.
|
|
|
|
``fan_out`` defaults to 1 and is never inferred. More than one machine means the same
|
|
work done several times over and the first verified result winning, which is worth
|
|
having for cheap idempotent work and wasteful for anything that costs GPU-seconds.
|
|
"""
|
|
|
|
node_ids: list[UUID] | None = None
|
|
fan_out: int = Field(default=1, ge=1, le=16)
|
|
lease_seconds: int = Field(default=60, ge=10, le=3600)
|
|
|
|
|
|
class ArtifactSpec(RequestModel):
|
|
"""One file a node says it produced.
|
|
|
|
Every bound here mirrors a database constraint, deliberately. ``content_hash`` is 64
|
|
lowercase hex characters because `artifacts_content_hash_check` says so, ``kind`` is
|
|
64 because `artifacts_kind_check` says so, and ``status`` is an enum because
|
|
`artifacts_status_check` says so. Matching them here rather than discovering them at
|
|
the INSERT is the difference between a node getting a 400 it can act on and a 500 it
|
|
cannot.
|
|
|
|
``content_hash`` and ``storage_key`` are both required, and an artifact without them
|
|
is not stored at all - verification would otherwise be checking the node's report
|
|
against the same node's report, which is not a check.
|
|
|
|
There is no object storage behind this yet (DECISION-6 is unresolved), so what is
|
|
recorded is the description of a file and where it is meant to be. That is enough to
|
|
establish that a node did the work, and not enough to pretend the bytes are
|
|
retrievable, so nothing here treats a ``storage_key`` as proof that it is.
|
|
"""
|
|
|
|
kind: str = Field(min_length=1, max_length=64)
|
|
content_hash: str = Field(pattern=r"^[0-9a-f]{64}$")
|
|
storage_key: str = Field(min_length=1, max_length=1024)
|
|
byte_size: int = Field(default=0, ge=0)
|
|
media_type: str = Field(default="application/octet-stream", max_length=200)
|
|
role: Literal["produced", "consumed"] = "produced"
|
|
"""`job_artifacts_role_check` allows exactly these two, and nothing else.
|
|
|
|
A node reporting what it made is `produced`. Worth being explicit that this is a
|
|
vocabulary the database fixed and the API had to learn rather than invent: the first
|
|
version of this field defaulted to `output`, which no constraint mentioned, so every
|
|
single completion would have failed with a check violation at the INSERT.
|
|
"""
|
|
lang: str | None = Field(default=None, max_length=16)
|
|
status: Literal["intermediate", "final", "superseded"] = "intermediate"
|
|
derivation: dict[str, Any] = Field(default_factory=dict)
|
|
|
|
|
|
class CompleteAssignmentRequest(RequestModel):
|
|
artifacts: list[ArtifactSpec] = Field(default_factory=list)
|
|
result: dict[str, Any] = Field(default_factory=dict)
|
|
|
|
|
|
class FailAssignmentRequest(RequestModel):
|
|
error: str = Field(min_length=1, max_length=2000)
|
|
retry: bool = True
|
|
|
|
|
|
@admin_router.post("/jobs/{job_id}/assign", response_model=list[AssignmentResponse])
|
|
def assign_job(
|
|
job_id: UUID,
|
|
payload: AssignJobRequest,
|
|
request: Request,
|
|
principal: Principal = Depends(require_admin),
|
|
) -> list[AssignmentResponse]:
|
|
"""Hand a job to one or more nodes.
|
|
|
|
An empty list is not an error: it means no machine was eligible, which is a fact an
|
|
operator needs to see rather than a 404. A job nobody can run is a queue that is not
|
|
draining, and it should say so.
|
|
"""
|
|
|
|
repository = compute_repository(request)
|
|
assignments = repository.assign_job(
|
|
job_id,
|
|
node_ids=payload.node_ids,
|
|
fan_out=payload.fan_out,
|
|
lease_seconds=payload.lease_seconds,
|
|
actor=getattr(principal, "email", None),
|
|
)
|
|
return [_assignment_response(one) for one in assignments]
|
|
|
|
|
|
@admin_router.post("/assignments/{assignment_id}/cancel", response_model=AssignmentOutcomeResponse)
|
|
def cancel_assignment(
|
|
assignment_id: UUID,
|
|
payload: ReasonRequest,
|
|
request: Request,
|
|
principal: Principal = Depends(require_admin),
|
|
) -> AssignmentOutcomeResponse:
|
|
"""Stop one attempt. The job itself carries on if anything else is open on it."""
|
|
|
|
repository = compute_repository(request)
|
|
outcome = repository.cancel_assignment(
|
|
assignment_id, reason=payload.reason, actor=getattr(principal, "email", None)
|
|
)
|
|
return _outcome_response(outcome)
|
|
|
|
|
|
@admin_router.get("/jobs/{job_id}/assignments", response_model=list[AssignmentResponse])
|
|
def list_job_assignments(
|
|
job_id: UUID,
|
|
request: Request,
|
|
principal: Principal = Depends(require_admin),
|
|
) -> list[AssignmentResponse]:
|
|
"""Every attempt made on this job, and which of them the controller verified.
|
|
|
|
This is the answer to "what was assigned, and what is not complete".
|
|
"""
|
|
|
|
repository = compute_repository(request)
|
|
return [_assignment_response(one) for one in repository.assignments_for_job(job_id)]
|
|
|
|
|
|
@agent_router.get("/assignments", response_model=list[AssignmentResponse])
|
|
def node_assignments(
|
|
request: Request,
|
|
node: NodeContext = Depends(require_node),
|
|
limit: int = Query(default=10, ge=1, le=100),
|
|
) -> list[AssignmentResponse]:
|
|
"""What this node has been asked to do.
|
|
|
|
Polling rather than pushed, and that is not a simplification. Nodes sit behind NAT and
|
|
the controller has no route to them; the only path that exists is a node calling out.
|
|
An agent that polls in a loop gets assignments as an ordinary consequence of running.
|
|
"""
|
|
|
|
node.require("compute:claim")
|
|
repository = compute_repository(request)
|
|
return [
|
|
_assignment_response(one)
|
|
for one in repository.assignments_for_node(node.node_id, limit=limit)
|
|
]
|
|
|
|
|
|
@agent_router.post("/assignments/{assignment_id}/start", response_model=AssignmentResponse)
|
|
def start_assignment(
|
|
assignment_id: UUID,
|
|
request: Request,
|
|
node: NodeContext = Depends(require_node),
|
|
) -> AssignmentResponse:
|
|
node.require("compute:claim")
|
|
repository = compute_repository(request)
|
|
return _assignment_response(repository.start_assignment(assignment_id, node.node_id))
|
|
|
|
|
|
@agent_router.post(
|
|
"/assignments/{assignment_id}/complete", response_model=AssignmentOutcomeResponse
|
|
)
|
|
def complete_assignment(
|
|
assignment_id: UUID,
|
|
payload: CompleteAssignmentRequest,
|
|
request: Request,
|
|
node: NodeContext = Depends(require_node),
|
|
) -> AssignmentOutcomeResponse:
|
|
"""Report the work finished, and hand over what was produced.
|
|
|
|
Reporting is not believing. The job only reaches ``complete`` if the controller can
|
|
point at artifacts, so a node that says it finished with nothing behind it gets its
|
|
assignment failed and the reason naming what was absent.
|
|
"""
|
|
|
|
node.require("compute:complete")
|
|
repository = compute_repository(request)
|
|
outcome = repository.complete_assignment(
|
|
assignment_id,
|
|
node.node_id,
|
|
artifacts=[spec.model_dump() for spec in payload.artifacts],
|
|
result=payload.result,
|
|
)
|
|
return _outcome_response(outcome)
|
|
|
|
|
|
@agent_router.post("/assignments/{assignment_id}/fail", response_model=AssignmentOutcomeResponse)
|
|
def fail_assignment(
|
|
assignment_id: UUID,
|
|
payload: FailAssignmentRequest,
|
|
request: Request,
|
|
node: NodeContext = Depends(require_node),
|
|
) -> AssignmentOutcomeResponse:
|
|
node.require("compute:complete")
|
|
repository = compute_repository(request)
|
|
outcome = repository.fail_assignment(
|
|
assignment_id, node.node_id, error=payload.error, retry=payload.retry
|
|
)
|
|
return _outcome_response(outcome)
|
|
|
|
|
|
class ExternalArtifactSpec(RequestModel):
|
|
"""One file that already exists somewhere, declared rather than uploaded.
|
|
|
|
Every bound mirrors a database constraint, for the same reason as `ArtifactSpec`:
|
|
a value the INSERT rejects is a 500 a caller cannot act on.
|
|
|
|
``storage_uri`` is required and is not fetched. It says where the bytes are; nothing
|
|
here goes and gets them, because there is no object storage to get them from
|
|
(DECISION-6 is unresolved). What the row therefore records is that a file is claimed to
|
|
exist at that address - see ``integrity``, which is ``unverified`` on every row written
|
|
here and stays that way until something reads the bytes back and hashes them.
|
|
"""
|
|
|
|
kind: str = Field(min_length=1, max_length=64)
|
|
content_hash: str = Field(pattern=r"^[0-9a-f]{64}$")
|
|
storage_uri: str = Field(min_length=1, max_length=2048)
|
|
storage_key: str | None = Field(default=None, max_length=1024)
|
|
byte_size: int = Field(default=0, ge=0)
|
|
media_type: str = Field(default="application/octet-stream", max_length=200)
|
|
status: Literal["intermediate", "final", "superseded"] = "intermediate"
|
|
lang: str | None = Field(default=None, max_length=16)
|
|
derivation: dict[str, Any] = Field(default_factory=dict)
|
|
|
|
|
|
class RegisterArtifactsRequest(RequestModel):
|
|
"""A manifest. One project's worth, declared in one call.
|
|
|
|
Bulk because a library is bulk: a finished audiobook is hundreds of files, and an
|
|
operator submitting them one at a time would abandon it partway and leave a catalogue
|
|
that is neither complete nor wrong - the worst state to be in.
|
|
"""
|
|
|
|
project_id: UUID
|
|
storage_node_id: UUID | None = None
|
|
artifacts: list[ExternalArtifactSpec] = Field(min_length=1, max_length=2000)
|
|
|
|
|
|
class RegisterArtifactsResponse(BaseModel):
|
|
"""What was added and what was already there.
|
|
|
|
``skipped`` names the entries rather than counting them. An operator who submitted a
|
|
300-file manifest and was told "done" would not know that two entries were malformed and
|
|
silently dropped; being able to see which two is the difference between a catalogue you
|
|
can rely on and one you cannot.
|
|
"""
|
|
|
|
added: list[str]
|
|
skipped: list[str]
|
|
total_added: int
|
|
|
|
|
|
class ArtifactResponse(BaseModel):
|
|
id: UUID
|
|
project_id: UUID
|
|
organization_id: UUID
|
|
kind: str
|
|
status: str
|
|
content_hash: str
|
|
byte_size: int
|
|
media_type: str
|
|
storage_key: str
|
|
storage_uri: str | None
|
|
storage_node_id: UUID | None
|
|
integrity: str
|
|
verified_at: datetime | None
|
|
lang: str | None
|
|
created_at: datetime | None
|
|
|
|
|
|
@admin_router.post("/artifacts", response_model=RegisterArtifactsResponse, status_code=201)
|
|
def register_artifacts(
|
|
payload: RegisterArtifactsRequest,
|
|
request: Request,
|
|
principal: Principal = Depends(require_admin),
|
|
organization_id: UUID | None = Query(default=None),
|
|
) -> RegisterArtifactsResponse:
|
|
"""Declare files that already exist, against a project in this organization.
|
|
|
|
The audio node is the master for the audiobooks; this is how its library enters the
|
|
system without the bytes moving. Idempotent on the file's hash within the project, so
|
|
the same manifest can be submitted twice - or by two operators who did not talk to each
|
|
other - without doubling the catalogue.
|
|
"""
|
|
|
|
repository = compute_repository(request)
|
|
target = organization_id or _organization_of(principal)
|
|
outcome = repository.register_artifacts(
|
|
target,
|
|
payload.project_id,
|
|
[spec.model_dump() for spec in payload.artifacts],
|
|
storage_node_id=payload.storage_node_id,
|
|
actor=getattr(principal, "email", None),
|
|
)
|
|
return RegisterArtifactsResponse(
|
|
added=outcome["added"],
|
|
skipped=outcome["skipped"],
|
|
total_added=len(outcome["added"]),
|
|
)
|
|
|
|
|
|
@admin_router.get("/artifacts", response_model=list[ArtifactResponse])
|
|
def list_artifacts(
|
|
request: Request,
|
|
principal: Principal = Depends(require_admin),
|
|
organization_id: UUID | None = Query(default=None),
|
|
project_id: UUID | None = Query(default=None),
|
|
kind: str | None = Query(default=None),
|
|
limit: int = Query(default=200, ge=1, le=1000),
|
|
) -> list[ArtifactResponse]:
|
|
"""The catalogue: what is claimed to exist, where, and whether anyone has checked.
|
|
|
|
Scoped to the caller's organization on every call. ``artifacts`` is platform-level, so
|
|
a missing predicate here would be one tenant's library readable by another.
|
|
"""
|
|
|
|
repository = compute_repository(request)
|
|
target = organization_id or _organization_of(principal)
|
|
rows = repository.list_artifacts(target, project_id=project_id, kind=kind, limit=limit)
|
|
return [
|
|
ArtifactResponse(
|
|
id=row["id"],
|
|
project_id=row["project_id"],
|
|
organization_id=row["organization_id"],
|
|
kind=row["kind"],
|
|
status=row["status"],
|
|
content_hash=row["content_hash"],
|
|
byte_size=int(row["byte_size"]),
|
|
media_type=row["media_type"],
|
|
storage_key=row["storage_key"],
|
|
storage_uri=row.get("storage_uri"),
|
|
storage_node_id=row.get("storage_node_id"),
|
|
integrity=row.get("integrity") or "unverified",
|
|
verified_at=row.get("verified_at"),
|
|
lang=row.get("lang"),
|
|
created_at=row.get("created_at"),
|
|
)
|
|
for row in rows
|
|
]
|
|
|
|
|
|
@admin_router.get("/models", response_model=list[ModelResponse])
|
|
def list_models(request: Request, principal: Principal = Depends(require_admin)) -> list[ModelResponse]:
|
|
"""The global model registry.
|
|
|
|
Global because "Mini-M4 can run qwen3-tts" is a fact about the model, not
|
|
about a tenant, and a per-tenant copy would let two organizations disagree
|
|
about what a model needs.
|
|
"""
|
|
|
|
repository = compute_repository(request)
|
|
return [ModelResponse(**row) for row in repository.list_models()]
|
|
|
|
|
|
@admin_router.post("/models", response_model=ModelResponse, status_code=201)
|
|
def register_model(
|
|
payload: ModelRequest,
|
|
request: Request,
|
|
principal: Principal = Depends(require_admin),
|
|
) -> ModelResponse:
|
|
"""Record what a model costs.
|
|
|
|
``vram_min_mb`` is the floor for running at all and ``vram_preferred_mb`` is
|
|
the point at which it runs well. Keeping them apart is what lets a machine
|
|
that can *just* manage the model take the job, slowly, instead of refusing
|
|
it. §7.3
|
|
"""
|
|
|
|
repository = compute_repository(request)
|
|
return ModelResponse(
|
|
**repository.register_model(
|
|
payload.slug,
|
|
kind=payload.kind,
|
|
vram_min_mb=payload.vram_min_mb,
|
|
vram_preferred_mb=payload.vram_preferred_mb,
|
|
ram_min_mb=payload.ram_min_mb,
|
|
requires_accelerators=payload.requires_accelerators,
|
|
reference_throughput=payload.reference_throughput,
|
|
load_seconds=payload.load_seconds,
|
|
)
|
|
)
|
|
|
|
|
|
@admin_router.post("/jobs", response_model=JobResponse, status_code=201)
|
|
def submit_job(
|
|
payload: SubmitJobRequest,
|
|
request: Request,
|
|
principal: Principal = Depends(require_admin),
|
|
organization_id: UUID | None = Query(default=None),
|
|
) -> JobResponse:
|
|
"""Add work.
|
|
|
|
With ``fail_if_unplaceable``, a job nothing can run is refused at submission
|
|
with the reason, rather than sitting in the queue looking like it is
|
|
progressing. That is the difference between "slow" and "impossible", and it
|
|
matters today because ``gemma4:26b`` can only run on one machine in the
|
|
inventory. §7.3.1
|
|
"""
|
|
|
|
repository = compute_repository(request)
|
|
target = organization_id or _organization_of(principal)
|
|
requirement = ModelRequirement(**payload.requires.model_dump())
|
|
job = repository.submit_job(
|
|
target,
|
|
kind=payload.kind,
|
|
requested_by=f"user:{principal.user_id}",
|
|
requires=requirement,
|
|
payload=payload.payload,
|
|
priority=payload.priority,
|
|
project_id=payload.project_id,
|
|
lang=payload.lang,
|
|
idempotency_key=payload.idempotency_key,
|
|
max_attempts=payload.max_attempts,
|
|
)
|
|
if payload.fail_if_unplaceable and job.status == "queued":
|
|
placement = repository.place(requirement, organization_id=target)
|
|
if placement is None:
|
|
reason = _unplaceable_reason(requirement, repository, target)
|
|
return _job_response(repository.mark_blocked(job.id, reason))
|
|
return _job_response(job)
|
|
|
|
|
|
def _unplaceable_reason(
|
|
requirement: ModelRequirement, repository, organization_id: UUID
|
|
) -> str:
|
|
"""Say what is missing, in terms a person can act on.
|
|
|
|
"no node satisfies this" is true and useless. The useful sentence names the
|
|
requirement and the closest thing that exists, because the usual cause is a
|
|
model pinned to one machine that happens to be busy.
|
|
"""
|
|
|
|
models = {row["slug"]: row for row in repository.list_models()}
|
|
if requirement.models:
|
|
for slug in requirement.models:
|
|
model = models.get(slug)
|
|
if model is None:
|
|
return f"model '{slug}' is not in the registry"
|
|
if model["requires_accelerators"]:
|
|
return (
|
|
f"model '{slug}' requires accelerator(s) "
|
|
f"{', '.join(model['requires_accelerators'])}; no online node advertises them"
|
|
)
|
|
nodes = repository.list_nodes(organization_id)
|
|
if not nodes:
|
|
return "no node in this organization is online"
|
|
online = [node for node in nodes if node.status == "online"]
|
|
if not online:
|
|
return f"no node is online ({', '.join(sorted({node.status for node in nodes}))})"
|
|
best = max(
|
|
(device.vram_free_mb for node in online for device in node.devices if not device.is_busy),
|
|
default=0,
|
|
)
|
|
if requirement.vram_min_mb and best < requirement.vram_min_mb:
|
|
return (
|
|
f"needs {requirement.vram_min_mb} MiB of free VRAM; the most any online device has "
|
|
f"is {best} MiB"
|
|
)
|
|
return "no online device satisfies this requirement"
|
|
|
|
|
|
@admin_router.get("/jobs", response_model=list[JobResponse])
|
|
def list_jobs(
|
|
request: Request,
|
|
principal: Principal = Depends(require_admin),
|
|
organization_id: UUID | None = Query(default=None),
|
|
status: list[str] | None = Query(default=None),
|
|
limit: int = Query(default=MAX_PAGE_SIZE, ge=1, le=MAX_PAGE_SIZE),
|
|
) -> list[JobResponse]:
|
|
"""The master view: what work exists, where it is running, and its status.
|
|
|
|
A projection over ``jobs`` rather than a status table of its own. A separate
|
|
table would be a cache of a truth that already exists, and it would drift
|
|
the way every other cache in this system has drifted. §7.5
|
|
"""
|
|
|
|
repository = compute_repository(request)
|
|
target = organization_id or _organization_of(principal)
|
|
return [_job_response(job) for job in repository.list_jobs(target, statuses=status)[:limit]]
|
|
|
|
|
|
@admin_router.get("/jobs/{job_id}/events", response_model=list[dict[str, Any]])
|
|
def job_events(
|
|
job_id: UUID,
|
|
request: Request,
|
|
principal: Principal = Depends(require_admin),
|
|
) -> list[dict[str, Any]]:
|
|
repository = compute_repository(request)
|
|
return [
|
|
{**event, "at": event["at"].isoformat() if event.get("at") else None}
|
|
for event in repository.job_events(job_id)
|
|
]
|
|
|
|
|
|
@admin_router.post("/leases/sweep", response_model=list[UUID])
|
|
def sweep_leases(
|
|
request: Request,
|
|
principal: Principal = Depends(require_admin),
|
|
) -> list[UUID]:
|
|
"""Return jobs whose holder stopped renewing.
|
|
|
|
Called by the controller's own timer. Exposed here so it can also be driven
|
|
by hand, because the first time you watch a killed agent's job come back is
|
|
the moment you believe the mechanism works. (I26)
|
|
"""
|
|
|
|
repository = compute_repository(request)
|
|
return repository.sweep_expired_leases()
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
|
|
@admin_router.post("/pairing-script")
|
|
def create_bootstrap(
|
|
payload: BootstrapRequest,
|
|
request: Request,
|
|
principal: Principal = Depends(require_admin),
|
|
organization_id: UUID | None = Query(default=None),
|
|
) -> BootstrapResponse:
|
|
"""Render a bootstrap script, and the code it will use.
|
|
|
|
Returns JSON rather than the bare script, because there are two callers with two
|
|
different needs. A person at a terminal wants a script they can pipe to `sh`; the
|
|
console wants the script *and* the code and the expiry, so it can show all three
|
|
once and then forget them. Returning text would force the console to parse a shell
|
|
script to recover a six-digit number, which is the kind of thing that works until
|
|
someone reformats the script.
|
|
|
|
The code appears in this response and in no later one - the controller keeps only an
|
|
Argon2id hash. That is why the console shows it once and offers no way to fetch it
|
|
again, rather than that being a UI choice.
|
|
"""
|
|
|
|
repository = compute_repository(request)
|
|
target = organization_id or _organization_of(principal)
|
|
minted = repository.create_pairing_code(
|
|
target,
|
|
payload.node_name,
|
|
locality=payload.locality,
|
|
trust_domain=payload.trust_domain,
|
|
ttl_seconds=payload.ttl_seconds,
|
|
)
|
|
controller = payload.controller_url.rstrip("/")
|
|
script = render_bootstrap(
|
|
node_name=payload.node_name,
|
|
pairing_code=minted["code"],
|
|
controller=controller,
|
|
as_root=payload.as_root,
|
|
)
|
|
return BootstrapResponse(
|
|
node_name=payload.node_name,
|
|
code=minted["code"],
|
|
code_hint=minted["code_hint"],
|
|
expires_at=minted["expires_at"],
|
|
script=script,
|
|
# The one line a person pastes. Built here rather than in the browser so the
|
|
# console and a shell user get the same command.
|
|
# The command follows `as_root`, and this was a real defect.
|
|
#
|
|
# `as_root` chooses which paths the script renders: /opt/umbra and
|
|
# /etc/umbra/node.json, or the user's own directories. The command did not follow
|
|
# it, so ticking the box produced a script that wanted to write to /etc while
|
|
# being run by an unprivileged shell - which fails at the first mkdir, after the
|
|
# pairing code has already been spent. A checkbox that produces a command which
|
|
# cannot work is worse than no checkbox.
|
|
command=(
|
|
f"curl -fsSL {controller}/api/node/bootstrap | "
|
|
+ ("sudo sh" if payload.as_root else "sh")
|
|
+ f"\n# then enter the code when asked: {minted['code']}"
|
|
),
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Agent surface: /api/node
|
|
#
|
|
# Everything below authenticates from a node credential, never from a session. A node
|
|
# has no browser, no cookie and no password, so anything that assumes one does not work
|
|
# here. Two routes are deliberately unauthenticated and are named as exceptions in
|
|
# test_compute.py with the reason each is safe.
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@agent_router.get("/agent")
|
|
def fetch_agent(node: NodeContext = Depends(require_node)) -> Response:
|
|
"""Serve the node agent to a credentialed node.
|
|
|
|
A code-distribution channel, so it is worth being precise about what it does and
|
|
does not add. A node credential already implies "run what the controller sends you"
|
|
- that is what claiming and completing a job is - so serving the agent grants no new
|
|
authority. What it removes is the courier: until now every node's agent was copied
|
|
onto it by hand, and the update path took a local file rather than a URL.
|
|
|
|
Served only to a node credential, never unauthenticated, because the agent file
|
|
reveals the protocol and the set of job kinds the controller speaks.
|
|
|
|
`X-Agent-Version` carries the version, and it is the same string the agent reports
|
|
in its heartbeat. A node can therefore ask what it *should* be running and compare,
|
|
rather than trusting whatever it currently has.
|
|
"""
|
|
|
|
node.require("compute:read")
|
|
source = Path(__file__).with_name("node_agent.py")
|
|
body = source.read_bytes()
|
|
version = ""
|
|
for line in body.decode("utf-8", "replace").splitlines():
|
|
if line.startswith("AGENT_VERSION"):
|
|
version = line.split("=", 1)[1].strip().strip("\"'")
|
|
break
|
|
return Response(
|
|
content=body,
|
|
media_type="text/x-python",
|
|
headers={
|
|
"X-Agent-Version": version or "unknown",
|
|
"Cache-Control": "no-store",
|
|
# The agent is source code the node will execute. It must not be cached by
|
|
# an intermediary, and it must not be sniffed into a different type.
|
|
"X-Content-Type-Options": "nosniff",
|
|
"Content-Disposition": 'attachment; filename="node_agent.py"',
|
|
},
|
|
)
|
|
|
|
|
|
@agent_router.get("/bootstrap")
|
|
def fetch_bootstrap(request: Request) -> Response:
|
|
"""Serve the bootstrap script itself.
|
|
|
|
The script embeds a pairing code, so it cannot be rendered without one - and this
|
|
route renders none. It serves the *template*: a script with placeholders and no
|
|
code, which is inert until an administrator fills it in. That is what makes
|
|
`curl ... | sh` a safe thing to show someone: the pipe downloads a program, and the
|
|
code that lets it do anything is typed by the operator afterwards.
|
|
|
|
Deliberately unauthenticated, and that is safe *only* because of what it returns. A
|
|
script with no code can check the machine, report what it finds, and stop. It cannot
|
|
pair, cannot authenticate, and cannot read anything.
|
|
|
|
Listed by name in the test that pins unauthenticated node routes, with its reason,
|
|
so a second open route has to be added deliberately.
|
|
"""
|
|
|
|
template = render_bootstrap_template(default_controller=_own_base_url(request))
|
|
return Response(
|
|
content=template,
|
|
media_type="text/x-sh",
|
|
headers={
|
|
"Cache-Control": "no-store",
|
|
"X-Content-Type-Options": "nosniff",
|
|
"X-Bootstrap-Version": bootstrap_version(),
|
|
},
|
|
)
|
|
|
|
|
|
@agent_router.post("/pair", response_model=NodePairResponse)
|
|
def pair_node(payload: NodePairRequest, request: Request) -> NodePairResponse:
|
|
"""Exchange a pairing code for a node credential.
|
|
|
|
Unauthenticated by design - that is what makes it useful, since the caller has
|
|
nothing else. Everything that makes it safe is in what it checks: a single-use
|
|
code, hashed at rest and expiring in minutes, bound to a node name the caller must
|
|
also match, and a per-address attempt budget taken before any code is even looked
|
|
up.
|
|
|
|
A wrong or spent code gives the same answer as a code that never existed, so the
|
|
response tells an attacker nothing per attempt. A code issued for a *different* node
|
|
says so plainly: that is not a secret, and an operator would otherwise be left
|
|
guessing whether they mistyped it.
|
|
|
|
Returns 403 rather than 401 on a bad code. Both are "no", and the distinction here
|
|
would only tell a prober which half of the guess was wrong.
|
|
"""
|
|
|
|
repository = compute_repository(request)
|
|
client = request.client
|
|
result = repository.redeem_pairing_code(
|
|
payload.code,
|
|
payload.node_name,
|
|
agent_version=payload.agent_version,
|
|
locality=payload.locality,
|
|
source_address=client.host if client else None,
|
|
mac=payload.mac,
|
|
)
|
|
if not result["ok"]:
|
|
raise PairingError(code=result["code"], message=result["message"])
|
|
return NodePairResponse(
|
|
node_id=result["node_id"],
|
|
node_name=result["node_name"],
|
|
base_name=result.get("base_name"),
|
|
key_prefix=result["key_prefix"],
|
|
key=result["key"],
|
|
scopes=result["scopes"],
|
|
)
|
|
|
|
|
|
@agent_router.post("/heartbeat", response_model=HeartbeatResponse)
|
|
def heartbeat(
|
|
payload: HeartbeatRequest,
|
|
request: Request,
|
|
node: NodeContext = Depends(require_node),
|
|
) -> HeartbeatResponse:
|
|
"""Report liveness and measurements. This is what makes a node ``online``.
|
|
|
|
The promotion happens here and nowhere else. A node an administrator has set
|
|
to ``draining`` or ``retired`` is not promoted back by heartbeating, so a
|
|
draining node that has not yet rebooted does not re-advertise itself for
|
|
work the administrator is trying to drain away from it. (I24)
|
|
"""
|
|
|
|
node.require("compute:heartbeat")
|
|
repository = compute_repository(request)
|
|
updated = repository.heartbeat(
|
|
node.node_id,
|
|
agent_version=payload.agent_version,
|
|
measurements=payload.measurements,
|
|
resident_models=payload.resident_models,
|
|
active_jobs=payload.active_jobs,
|
|
devices=[device.model_dump() for device in payload.devices],
|
|
)
|
|
return HeartbeatResponse(
|
|
node_id=updated.id,
|
|
name=updated.name,
|
|
status=updated.status,
|
|
schedulable=updated.is_schedulable,
|
|
status_reason=_node_reason(updated),
|
|
devices=[_device_dict(device) for device in updated.devices],
|
|
)
|
|
|
|
|
|
@agent_router.post("/models/{slug}/state", response_model=dict[str, Any])
|
|
def set_model_state(
|
|
slug: str,
|
|
payload: dict[str, Any],
|
|
request: Request,
|
|
node: NodeContext = Depends(require_node),
|
|
) -> dict[str, Any]:
|
|
"""Record what this node has resident.
|
|
|
|
The input to the residency half of placement scoring, and the reason a node
|
|
at 0% utilisation with 87% of its VRAM held shows as constrained rather than
|
|
as idle. §7.3
|
|
"""
|
|
|
|
node.require("compute:heartbeat")
|
|
repository = compute_repository(request)
|
|
repository.set_node_model_state(
|
|
node.node_id,
|
|
slug,
|
|
payload.get("state", "loaded"),
|
|
device_id=payload.get("device_id"),
|
|
vram_held_mb=int(payload.get("vram_held_mb", 0)),
|
|
)
|
|
return {"ok": True}
|
|
|
|
|
|
@agent_router.post("/jobs/claim", response_model=ClaimedJobResponse | None)
|
|
def claim_job(
|
|
request: Request,
|
|
node: NodeContext = Depends(require_node),
|
|
kinds: list[str] | None = Query(default=None),
|
|
lease_seconds: int = Query(default=DEFAULT_LEASE_SECONDS, ge=10, le=3600),
|
|
) -> ClaimedJobResponse | None:
|
|
"""Take one job, atomically, or get nothing.
|
|
|
|
``FOR UPDATE SKIP LOCKED`` inside the claim is what lets several agents take
|
|
several *different* jobs with no coordination. Without it two of them pick
|
|
the same row and one overwrites the other's claim, and the result is
|
|
duplicated work that is individually plausible and collectively wrong. (I25)
|
|
|
|
A ``null`` body means there was nothing to take, which is the normal case and
|
|
not an error.
|
|
"""
|
|
|
|
node.require("compute:claim")
|
|
repository = compute_repository(request)
|
|
claimed = repository.claim_job(
|
|
node.node_id, lease_seconds=lease_seconds, kinds=kinds
|
|
)
|
|
if claimed is None:
|
|
return None
|
|
return ClaimedJobResponse(
|
|
job=_job_response(claimed.job),
|
|
placement=PlacementResponse(**claimed.placement.to_json()),
|
|
lease_expires_at=claimed.lease_expires_at.isoformat(),
|
|
renew_after_seconds=max(lease_seconds // 3, 5),
|
|
)
|
|
|
|
|
|
@agent_router.post("/jobs/{job_id}/renew", response_model=RenewResponse)
|
|
def renew_lease(
|
|
job_id: UUID,
|
|
request: Request,
|
|
node: NodeContext = Depends(require_node),
|
|
lease_seconds: int = Query(default=DEFAULT_LEASE_SECONDS, ge=10, le=3600),
|
|
) -> RenewResponse:
|
|
"""Extend a lease while the work continues.
|
|
|
|
A lease is a liveness signal, not a timeout for the work. A long render
|
|
renews every few seconds; the lease expires only when the node actually
|
|
stops renewing, which is when the job is genuinely lost. §6.17
|
|
"""
|
|
|
|
node.require("compute:claim")
|
|
repository = compute_repository(request)
|
|
expires = repository.renew_lease(job_id, node.node_id, lease_seconds=lease_seconds)
|
|
return RenewResponse(
|
|
job_id=job_id,
|
|
lease_expires_at=expires.isoformat() if expires else None,
|
|
ok=expires is not None,
|
|
)
|
|
|
|
|
|
@agent_router.post("/jobs/{job_id}/complete", response_model=JobResponse)
|
|
def complete_job(
|
|
job_id: UUID,
|
|
payload: CompleteJobRequest,
|
|
request: Request,
|
|
node: NodeContext = Depends(require_node),
|
|
) -> JobResponse:
|
|
node.require("compute:complete")
|
|
repository = compute_repository(request)
|
|
return _job_response(repository.complete_job(job_id, node.node_id, result=payload.result))
|
|
|
|
|
|
@agent_router.post("/jobs/{job_id}/fail", response_model=JobResponse)
|
|
def fail_job(
|
|
job_id: UUID,
|
|
payload: FailJobRequest,
|
|
request: Request,
|
|
node: NodeContext = Depends(require_node),
|
|
) -> JobResponse:
|
|
"""Record a failure and decide whether the job gets another attempt.
|
|
|
|
A released lease is *not* a failure and does not come through here - that is
|
|
the sweep. This path is for a job the node tried and could not do, and it
|
|
ends ``failed`` rather than retrying forever once ``max_attempts`` is spent.
|
|
"""
|
|
|
|
node.require("compute:complete")
|
|
repository = compute_repository(request)
|
|
return _job_response(
|
|
repository.fail_job(job_id, node.node_id, error=payload.error, retry=payload.retry)
|
|
)
|
|
|
|
|
|
@agent_router.post("/jobs/{job_id}/decompose", response_model=list[JobResponse])
|
|
def decompose(
|
|
job_id: UUID,
|
|
payload: DecomposeRequest,
|
|
request: Request,
|
|
node: NodeContext = Depends(require_node),
|
|
) -> list[JobResponse]:
|
|
"""Hand a job's pieces back for allocation.
|
|
|
|
Children are ordinary jobs with a ``parent_job_id``, so the scheduler has no
|
|
special case and the queue has no second table. The parent becomes a
|
|
container: released, never itself claimed. A parent with any failed child
|
|
ends ``partial``, never ``complete`` - "3 of 49 chapters analysed" and
|
|
"failed" are different facts. (I29, I30)
|
|
"""
|
|
|
|
node.require("compute:complete")
|
|
repository = compute_repository(request)
|
|
children = repository.submit_children(
|
|
job_id,
|
|
node.node_id,
|
|
[child.model_dump() for child in payload.children],
|
|
)
|
|
return [_job_response(child) for child in children]
|
|
|
|
|
|
class AudioIngestRequest(RequestModel):
|
|
"""An audio dataset bundle in the accepted format.
|
|
|
|
The audio side may POST the full envelope and records (same structure as
|
|
bundle.json + manifest.jsonl), allowing the server to validate and map
|
|
without fetching files.
|
|
"""
|
|
|
|
envelope: dict[str, Any]
|
|
records: list[dict[str, Any]]
|
|
project_name: str | None = Field(default=None, max_length=200)
|
|
project_slug: str | None = Field(default=None, max_length=200)
|
|
node_id: str | None = Field(default=None, max_length=200)
|
|
dry_run: bool = True
|
|
confirm: str | None = Field(default=None, max_length=128)
|
|
|
|
|
|
class AudioIngestWarningResponse(BaseModel):
|
|
code: str
|
|
subject: str
|
|
message: str
|
|
|
|
|
|
class AudioIngestCountsResponse(BaseModel):
|
|
chapters: int
|
|
speakers: int
|
|
utterances: int
|
|
artifacts: int
|
|
warnings: int
|
|
|
|
|
|
class AudioIngestResultResponse(BaseModel):
|
|
project_id: UUID | None = None
|
|
import_id: UUID | None = None
|
|
external_id: str | None = None
|
|
content_hash: str | None = None
|
|
counts: AudioIngestCountsResponse | None = None
|
|
imported_at: str | None = None
|
|
|
|
|
|
class AudioIngestResponse(BaseModel):
|
|
status: str
|
|
content_hash: str
|
|
narrator_external_id: str | None = None
|
|
counts: AudioIngestCountsResponse
|
|
warnings: list[AudioIngestWarningResponse]
|
|
result: AudioIngestResultResponse | None = None
|
|
|
|
|
|
@agent_router.post("/audio/import", response_model=AudioIngestResponse)
|
|
def import_audio_dataset(
|
|
payload: AudioIngestRequest,
|
|
request: Request,
|
|
node: NodeContext = Depends(require_node),
|
|
) -> AudioIngestResponse:
|
|
"""Import an audio bundle. Dry-runs by default; commit requires confirm."""
|
|
|
|
node.require("dataset:ingest")
|
|
|
|
from .audio_import import (
|
|
Bundle,
|
|
BundleError,
|
|
InMemoryAudioImportStore,
|
|
import_bundle,
|
|
)
|
|
|
|
try:
|
|
declared = payload.envelope.get("content_hash")
|
|
declared_value = (
|
|
declared.get("value") if isinstance(declared, dict) else None
|
|
)
|
|
manifest_bytes = (
|
|
"\n".join(
|
|
json.dumps(
|
|
r,
|
|
sort_keys=True,
|
|
separators=(",", ":"),
|
|
ensure_ascii=False,
|
|
)
|
|
for r in payload.records
|
|
)
|
|
+ "\n"
|
|
)
|
|
manifest_sha256 = hashlib.sha256(manifest_bytes.encode("utf-8")).hexdigest()
|
|
bundle = Bundle(
|
|
path=Path("."),
|
|
envelope=payload.envelope,
|
|
records=tuple(payload.records),
|
|
manifest_sha256=manifest_sha256,
|
|
declared_content_hash=declared_value
|
|
if isinstance(declared_value, str)
|
|
else None,
|
|
)
|
|
except Exception as error:
|
|
raise BundleError(f"invalid bundle payload: {error}")
|
|
|
|
# A dry run is pure validation and never touches the database. A commit goes
|
|
# through the PostgreSQL adapter or fails; there is deliberately no in-memory
|
|
# commit path, because a commit that writes nothing and reports success is
|
|
# the failure this route must not reproduce.
|
|
#
|
|
# The organization comes from the credential, never from the payload: a node
|
|
# belongs to exactly one organization, and letting a request name one would
|
|
# let a node write into another tenant's tables.
|
|
store = (
|
|
InMemoryAudioImportStore() if payload.dry_run else audio_import_store(request)
|
|
)
|
|
try:
|
|
outcome = import_bundle(
|
|
bundle,
|
|
store,
|
|
organization_id=node.organization_id,
|
|
project_name=payload.project_name,
|
|
project_slug=payload.project_slug,
|
|
node_id=payload.node_id or node.node_id,
|
|
dry_run=payload.dry_run,
|
|
confirm=payload.confirm,
|
|
)
|
|
except BundleError:
|
|
raise
|
|
except Exception as error:
|
|
raise BundleError(str(error))
|
|
|
|
result = outcome.result
|
|
counts = AudioIngestCountsResponse(
|
|
chapters=len(outcome.plan.chapters),
|
|
speakers=len(outcome.plan.speakers),
|
|
utterances=len(outcome.plan.utterances),
|
|
artifacts=len(outcome.plan.artifacts),
|
|
warnings=len(outcome.plan.warnings),
|
|
)
|
|
warnings = [
|
|
AudioIngestWarningResponse(
|
|
code=w.code, subject=w.subject, message=w.message
|
|
)
|
|
for w in outcome.plan.warnings
|
|
]
|
|
res = AudioIngestResultResponse()
|
|
if result is not None:
|
|
res = AudioIngestResultResponse(
|
|
project_id=result.project_id,
|
|
import_id=result.import_id,
|
|
external_id=result.external_id,
|
|
content_hash=result.content_hash,
|
|
counts=AudioIngestCountsResponse(**result.counts),
|
|
imported_at=result.imported_at.isoformat(),
|
|
)
|
|
return AudioIngestResponse(
|
|
status=outcome.status,
|
|
content_hash=outcome.plan.project.content_hash,
|
|
narrator_external_id=outcome.plan.narrator_external_id,
|
|
counts=counts,
|
|
warnings=warnings,
|
|
result=res if (outcome.result is not None or outcome.status == "no_change") else None,
|
|
)
|
|
|
|
|
|
__all__ = ["admin_router", "agent_router"]
|
|
|
|
|
|
def _own_base_url(request: Request) -> str:
|
|
"""The address this service is reachable at, as the caller used it.
|
|
|
|
Taken from the Host header rather than a configured value, because the whole point
|
|
is that the template works on whichever deployment served it. A hardcoded controller
|
|
address was the first version and it meant every other deployment got a script that
|
|
could not resolve its own controller.
|
|
"""
|
|
|
|
host = request.headers.get("host") or ""
|
|
scheme = request.headers.get("x-forwarded-proto") or request.url.scheme or "http"
|
|
if not host:
|
|
return ""
|
|
return f"{scheme}://{host}"
|