初始化 HyperNode-AI 控制面。

包含节点账本、每日清算，以及 Ubuntu 交互式部署脚本。 Co-authored-by: Cursor <cursoragent@cursor.com>
2026-10-10 21:53:20 +08:00
commit 2bbdb00a4d
31 changed files with 2450 additions and 0 deletions
@@ -0,0 +1 @@
+"""业务服务。"""
@@ -0,0 +1,21 @@
+"""按序列号或钱包地址读取主体，供各业务模块共用。"""
+
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+
+from app.errors import AppError
+from app.models import ComputeNode, User
+
+
+def get_user_by_wallet(session: Session, wallet_address: str) -> User:
+    user = session.scalar(select(User).where(User.wallet_address == wallet_address))
+    if user is None:
+        raise AppError("用户不存在", 404)
+    return user
+
+
+def get_node_by_sn(session: Session, sn: str) -> ComputeNode:
+    node = session.scalar(select(ComputeNode).where(ComputeNode.sn == sn))
+    if node is None:
+        raise AppError(f"节点不存在: {sn}", 404)
+    return node
@@ -0,0 +1,78 @@
+"""心跳登记与离线巡检。"""
+
+from __future__ import annotations
+
+from datetime import datetime, timedelta
+
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+
+from app.clock import utcnow
+from app.config import settings
+from app.errors import AppError
+from app.models import ComputeNode, NodeStatus, UptimeInterval
+from app.services.common import get_node_by_sn
+
+
+def record_heartbeat(session: Session, sn: str, now: datetime | None = None) -> ComputeNode:
+    """刷新最后活跃时间。离线节点收到心跳后重新打开一段在线区间。"""
+    moment = now or utcnow()
+    node = get_node_by_sn(session, sn.strip())
+    if node.license_expires_at <= moment:
+        raise AppError("节点授权已到期", 403)
+
+    node.last_heartbeat_at = moment
+    node.updated_at = moment
+    if node.status != NodeStatus.ONLINE:
+        node.status = NodeStatus.ONLINE
+        session.add(UptimeInterval(node_id=node.id, started_at=moment))
+    else:
+        _ensure_open_interval(session, node.id, moment)
+    session.flush()
+    return node
+
+
+def mark_stale_nodes(session: Session, now: datetime | None = None) -> list[str]:
+    """把超过心跳时限或授权已到期的在线节点置为 offline，并关闭在线区间。"""
+    moment = now or utcnow()
+    cutoff = moment - timedelta(minutes=settings.heartbeat_timeout_minutes)
+    nodes = session.scalars(select(ComputeNode).where(ComputeNode.status == NodeStatus.ONLINE)).all()
+    changed: list[str] = []
+    for node in nodes:
+        reasons: list[tuple[datetime, str]] = []
+        if node.last_heartbeat_at is None or node.last_heartbeat_at < cutoff:
+            reasons.append((node.last_heartbeat_at or moment, "heartbeat_timeout"))
+        if node.license_expires_at <= moment:
+            reasons.append((node.license_expires_at, "license_expired"))
+        if not reasons:
+            continue
+        ended_at, reason = min(reasons, key=lambda item: item[0])
+        _close_open_intervals(session, node.id, ended_at, reason)
+        node.status = NodeStatus.OFFLINE
+        node.updated_at = moment
+        changed.append(node.sn)
+    session.flush()
+    return changed
+
+
+def _ensure_open_interval(session: Session, node_id: int, started_at: datetime) -> None:
+    opened = session.scalar(
+        select(UptimeInterval).where(
+            UptimeInterval.node_id == node_id,
+            UptimeInterval.ended_at.is_(None),
+        )
+    )
+    if opened is None:
+        session.add(UptimeInterval(node_id=node_id, started_at=started_at))
+
+
+def _close_open_intervals(session: Session, node_id: int, ended_at: datetime, reason: str) -> None:
+    rows = session.scalars(
+        select(UptimeInterval).where(
+            UptimeInterval.node_id == node_id,
+            UptimeInterval.ended_at.is_(None),
+        )
+    ).all()
+    for row in rows:
+        row.ended_at = ended_at if ended_at >= row.started_at else row.started_at
+        row.close_reason = reason
@@ -0,0 +1,41 @@
+"""多实例定时任务互斥。锁有过期时间，持有进程异常退出后其他实例仍能继续。"""
+
+from __future__ import annotations
+
+import os
+import socket
+from datetime import timedelta
+
+from sqlalchemy import update
+from sqlalchemy.exc import IntegrityError
+
+from app.clock import utcnow
+from app.database import session_scope
+from app.models import SchedulerLock
+
+OWNER = f"{socket.gethostname()}:{os.getpid()}"
+
+
+def acquire_lock(name: str, ttl: timedelta) -> bool:
+    now = utcnow()
+    until = now + ttl
+    try:
+        with session_scope() as session:
+            lock = session.get(SchedulerLock, name)
+            if lock is None:
+                session.add(SchedulerLock(name=name, locked_until=until, owner=OWNER))
+                try:
+                    session.flush()
+                except IntegrityError:
+                    session.rollback()
+                    return False
+                return True
+            updated = session.execute(
+                update(SchedulerLock)
+                .where(SchedulerLock.name == name)
+                .where(SchedulerLock.locked_until <= now)
+                .values(locked_until=until, owner=OWNER)
+            )
+            return updated.rowcount == 1
+    except IntegrityError:
+        return False
@@ -0,0 +1,74 @@
+"""用户与节点登记。"""
+
+from __future__ import annotations
+
+from decimal import Decimal
+from urllib.parse import urlparse
+
+from sqlalchemy import select
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.orm import Session
+
+from app.clock import add_months, utcnow
+from app.config import settings
+from app.errors import AppError
+from app.models import ComputeNode, NodeStatus, User
+from app.money import money
+from app.services.common import get_user_by_wallet
+
+
+def register_user(session: Session, wallet_address: str) -> User:
+    wallet = wallet_address.strip()
+    if session.scalar(select(User).where(User.wallet_address == wallet)) is not None:
+        raise AppError("钱包地址已注册", 409)
+    user = User(wallet_address=wallet, created_at=utcnow())
+    session.add(user)
+    try:
+        session.flush()
+    except IntegrityError as exc:
+        session.rollback()
+        raise AppError("钱包地址已注册", 409) from exc
+    return user
+
+
+def register_node(
+    session: Session,
+    wallet_address: str,
+    sn: str,
+    purchase_amount: Decimal,
+    callback_url: str | None = None,
+) -> ComputeNode:
+    user = get_user_by_wallet(session, wallet_address.strip())
+    serial = sn.strip()
+    if session.scalar(select(ComputeNode).where(ComputeNode.sn == serial)) is not None:
+        raise AppError("节点序列号已存在", 409)
+    callback = _normalize_callback(callback_url)
+    started = utcnow()
+    node = ComputeNode(
+        user_id=user.id,
+        sn=serial,
+        purchase_amount=money(purchase_amount),
+        license_months=settings.license_months,
+        licensed_at=started,
+        license_expires_at=add_months(started, settings.license_months),
+        status=NodeStatus.OFFLINE,
+        callback_url=callback,
+        created_at=started,
+        updated_at=started,
+    )
+    session.add(node)
+    try:
+        session.flush()
+    except IntegrityError as exc:
+        session.rollback()
+        raise AppError("节点序列号已存在", 409) from exc
+    return node
+
+
+def _normalize_callback(callback_url: str | None) -> str | None:
+    if callback_url is None or callback_url.strip() == "":
+        return None
+    parsed = urlparse(callback_url.strip())
+    if parsed.scheme not in {"http", "https"} or not parsed.netloc:
+        raise AppError("callback_url 必须是 http 或 https 地址")
+    return callback_url.strip()
@@ -0,0 +1,183 @@
+"""随机任务下发与回传校验。
+
+节点不暴露回调地址时，任务留在待拉取队列，客户端用
+GET /api/v1/node/{sn}/tasks 领取。配置了 callback_url 时，调度器会额外 POST 一次。
+回传必须带上任务里的 nonce，避免未收到任务的客户端直接上报成功。
+"""
+
+from __future__ import annotations
+
+import hmac
+import json
+import logging
+import random
+import secrets
+import uuid
+from datetime import datetime, timedelta
+from urllib.error import URLError
+from urllib.request import Request, urlopen
+
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+
+from app.clock import utcnow
+from app.config import settings
+from app.errors import AppError
+from app.models import ComputeNode, ComputeTask, NodeStatus, PushStatus, TaskStatus
+from app.services.common import get_node_by_sn
+
+logger = logging.getLogger(__name__)
+
+
+def dispatch_random_tasks(session: Session, now: datetime | None = None) -> list[ComputeTask]:
+    """向当前在线、授权仍有效、且没有未完成任务的节点随机下发一轮任务。"""
+    moment = now or utcnow()
+    expire_overdue_tasks(session, moment)
+    pending_node_ids = set(
+        session.scalars(select(ComputeTask.node_id).where(ComputeTask.status == TaskStatus.PENDING)).all()
+    )
+    candidates = [
+        node
+        for node in session.scalars(
+            select(ComputeNode).where(
+                ComputeNode.status == NodeStatus.ONLINE,
+                ComputeNode.license_expires_at > moment,
+            )
+        ).all()
+        if node.id not in pending_node_ids
+    ]
+    if not candidates:
+        return []
+
+    batch = min(len(candidates), settings.dispatch_batch_size)
+    chosen = random.sample(candidates, batch)
+    deadline = moment + timedelta(minutes=settings.task_deadline_minutes)
+    tasks: list[ComputeTask] = []
+    for node in chosen:
+        task = ComputeTask(
+            id=str(uuid.uuid4()),
+            node_id=node.id,
+            task_type="simulated_compute",
+            payload={
+                "op": "simulated_compute",
+                "nonce": secrets.token_hex(16),
+                "seed": secrets.token_hex(8),
+            },
+            status=TaskStatus.PENDING,
+            push_status=PushStatus.QUEUED,
+            dispatched_at=moment,
+            deadline_at=deadline,
+        )
+        session.add(task)
+        tasks.append(task)
+    session.flush()
+    for node, task in zip(chosen, tasks):
+        _push_task(node, task)
+    session.flush()
+    return tasks
+
+
+def list_pending_tasks(session: Session, sn: str, now: datetime | None = None) -> list[ComputeTask]:
+    moment = now or utcnow()
+    node = get_node_by_sn(session, sn.strip())
+    tasks = session.scalars(
+        select(ComputeTask)
+        .where(ComputeTask.node_id == node.id, ComputeTask.status == TaskStatus.PENDING)
+        .order_by(ComputeTask.dispatched_at)
+    ).all()
+    visible: list[ComputeTask] = []
+    for task in tasks:
+        if task.deadline_at < moment:
+            task.status = TaskStatus.EXPIRED
+            continue
+        visible.append(task)
+    session.flush()
+    return visible
+
+
+def report_task(
+    session: Session,
+    sn: str,
+    task_id: str,
+    success: bool,
+    nonce: str,
+    now: datetime | None = None,
+) -> ComputeTask:
+    """校验任务归属、时限和 nonce。只有校验通过且 success 为真才记为有效响应。"""
+    moment = now or utcnow()
+    node = get_node_by_sn(session, sn.strip())
+    task = session.get(ComputeTask, task_id)
+    if task is None:
+        raise AppError("任务不存在", 404)
+    if task.node_id != node.id:
+        raise AppError("任务不属于该节点", 403)
+    if task.status != TaskStatus.PENDING:
+        raise AppError("任务已结束，不能重复回传", 409)
+    if task.deadline_at < moment:
+        task.status = TaskStatus.EXPIRED
+        session.flush()
+        raise AppError("任务已超过回传时限", 409)
+
+    expected = str((task.payload or {}).get("nonce", ""))
+    if not _nonce_matches(nonce, expected):
+        raise AppError("任务响应校验失败")
+
+    task.responded_at = moment
+    task.success = success
+    task.status = TaskStatus.COMPLETED if success else TaskStatus.FAILED
+    session.flush()
+    return task
+
+
+def expire_overdue_tasks(session: Session, now: datetime | None = None) -> int:
+    moment = now or utcnow()
+    tasks = session.scalars(
+        select(ComputeTask).where(
+            ComputeTask.status == TaskStatus.PENDING,
+            ComputeTask.deadline_at < moment,
+        )
+    ).all()
+    for task in tasks:
+        task.status = TaskStatus.EXPIRED
+    if tasks:
+        session.flush()
+    return len(tasks)
+
+
+def _nonce_matches(provided: str, expected: str) -> bool:
+    if not expected or len(provided) != len(expected):
+        return False
+    return hmac.compare_digest(provided, expected)
+
+
+def _push_task(node: ComputeNode, task: ComputeTask) -> None:
+    if not node.callback_url:
+        task.push_status = PushStatus.QUEUED
+        return
+    body = json.dumps(
+        {
+            "task_id": task.id,
+            "sn": node.sn,
+            "task_type": task.task_type,
+            "payload": task.payload,
+            "deadline_at": task.deadline_at.isoformat(),
+        }
+    ).encode()
+    request = Request(
+        node.callback_url,
+        data=body,
+        headers={"Content-Type": "application/json"},
+        method="POST",
+    )
+    try:
+        with urlopen(request, timeout=2) as response:
+            if 200 <= response.status < 300:
+                task.push_status = PushStatus.PUSHED
+                task.push_error = None
+                return
+            task.push_status = PushStatus.FAILED
+            task.push_error = f"HTTP {response.status}"
+    except (URLError, TimeoutError, OSError) as exc:
+        task.push_status = PushStatus.FAILED
+        task.push_error = str(exc)[:255]
+        logger.warning("task push failed sn=%s task=%s error=%s", node.sn, task.id, exc)
@@ -0,0 +1,99 @@
+"""提现申请。可用余额 = 已入账积分 - 待处理提现 - 已完成提现。"""
+
+from __future__ import annotations
+
+from decimal import Decimal
+
+from sqlalchemy import select
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.orm import Session
+
+from app.clock import utcnow
+from app.errors import AppError
+from app.models import NodeYield, WithdrawalLog, WithdrawalStatus, YieldStatus
+from app.money import money, to_decimal
+from app.services.common import get_user_by_wallet
+
+
+def available_balance(session: Session, user_id: int) -> Decimal:
+    credited_rows = session.scalars(
+        select(NodeYield.points_delta).where(
+            NodeYield.user_id == user_id,
+            NodeYield.status == YieldStatus.SETTLED,
+        )
+    ).all()
+    reserved_rows = session.scalars(
+        select(WithdrawalLog.amount).where(
+            WithdrawalLog.user_id == user_id,
+            WithdrawalLog.status.in_([WithdrawalStatus.PENDING, WithdrawalStatus.COMPLETED]),
+        )
+    ).all()
+    credited = sum((to_decimal(item) for item in credited_rows), Decimal("0"))
+    reserved = sum((to_decimal(item) for item in reserved_rows), Decimal("0"))
+    balance = credited - reserved
+    if balance <= 0:
+        return Decimal("0")
+    return money(balance)
+
+
+def request_withdrawal(
+    session: Session,
+    wallet_address: str,
+    amount: Decimal,
+    asset_symbol: str = "POINT",
+) -> tuple[WithdrawalLog, Decimal]:
+    user = get_user_by_wallet(session, wallet_address.strip())
+    value = money(amount)
+    if value <= 0:
+        raise AppError("提现金额必须大于 0")
+    balance = available_balance(session, user.id)
+    if value > balance:
+        raise AppError("可用积分不足")
+    row = WithdrawalLog(
+        user_id=user.id,
+        amount=value,
+        asset_symbol=asset_symbol.strip().upper() or "POINT",
+        status=WithdrawalStatus.PENDING,
+        created_at=utcnow(),
+        updated_at=utcnow(),
+    )
+    session.add(row)
+    session.flush()
+    return row, money(balance - value)
+
+
+def confirm_withdrawal(session: Session, withdrawal_id: int, tx_hash: str) -> WithdrawalLog:
+    row = _get_withdrawal(session, withdrawal_id)
+    if row.status != WithdrawalStatus.PENDING:
+        raise AppError("仅待处理的提现可以确认", 409)
+    digest = tx_hash.strip()
+    occupied = session.scalar(select(WithdrawalLog.id).where(WithdrawalLog.tx_hash == digest))
+    if occupied is not None:
+        raise AppError("交易哈希已存在", 409)
+    row.tx_hash = digest
+    row.status = WithdrawalStatus.COMPLETED
+    row.updated_at = utcnow()
+    try:
+        session.flush()
+    except IntegrityError as exc:
+        session.rollback()
+        raise AppError("交易哈希已存在", 409) from exc
+    return row
+
+
+def reject_withdrawal(session: Session, withdrawal_id: int, note: str | None = None) -> WithdrawalLog:
+    row = _get_withdrawal(session, withdrawal_id)
+    if row.status != WithdrawalStatus.PENDING:
+        raise AppError("仅待处理的提现可以拒绝", 409)
+    row.status = WithdrawalStatus.REJECTED
+    row.note = note
+    row.updated_at = utcnow()
+    session.flush()
+    return row
+
+
+def _get_withdrawal(session: Session, withdrawal_id: int) -> WithdrawalLog:
+    row = session.get(WithdrawalLog, withdrawal_id)
+    if row is None:
+        raise AppError("提现记录不存在", 404)
+    return row
@@ -0,0 +1,236 @@
+"""每日产出清算。
+
+释放规则：
+1. 授权金额在整个授权天数内线性释放。授权天数取 license 起止日期之差。
+2. 在线率 = 当日授权窗口内的在线秒数 / 当日授权有效秒数。
+   尚未关闭的在线区间，若最后心跳已超过超时时间，只计算到最后一次心跳。
+3. 任务响应率 = 校验通过且执行成功的任务数 / 当日内下发任务数。
+4. 当天没有下发任务时，表现系数等于在线率，避免调度器故障误伤节点。
+   有任务时，表现系数 = 在线权重 × 在线率 + 任务权重 × 响应率。
+5. 算力产出 = 授权金额 × 表现系数。
+   当日积分 = 算力产出 / 授权天数。
+6. 单节点累计积分不超过授权金额。已存在的当日流水不会被覆盖。
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import date, datetime, timedelta
+from decimal import Decimal
+
+from sqlalchemy import or_, select
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.orm import Session
+
+from app.clock import day_bounds, utcnow
+from app.config import settings
+from app.errors import AppError
+from app.models import ComputeNode, ComputeTask, NodeYield, TaskStatus, UptimeInterval, YieldStatus
+from app.money import money, rate, ratio, to_decimal
+from app.services.common import get_node_by_sn
+from app.services.tasks import expire_overdue_tasks
+
+
+@dataclass
+class YieldComputation:
+    eligible: bool
+    yield_date: date
+    online_seconds: int = 0
+    eligible_seconds: int = 0
+    online_rate: Decimal = Decimal("0")
+    tasks_dispatched: int = 0
+    tasks_completed: int = 0
+    task_response_rate: Decimal = Decimal("0")
+    online_weight: Decimal = Decimal("0")
+    task_weight: Decimal = Decimal("0")
+    performance: Decimal = Decimal("0")
+    compute_output: Decimal = Decimal("0")
+    points_delta: Decimal = Decimal("0")
+    cycle_days: int = 0
+
+
+@dataclass
+class SettleSummary:
+    yield_date: date
+    created: int
+    skipped: int
+    total_points: Decimal
+    is_partial: bool
+
+
+def compute_node_yield(
+    session: Session,
+    node: ComputeNode,
+    yield_date: date,
+    now: datetime | None = None,
+) -> YieldComputation:
+    moment = now or utcnow()
+    day_start, day_end = day_bounds(yield_date)
+    window_start = max(day_start, node.licensed_at)
+    window_end = min(day_end, node.license_expires_at)
+    cycle_days = (node.license_expires_at.date() - node.licensed_at.date()).days
+    if window_end <= window_start or cycle_days <= 0:
+        return YieldComputation(eligible=False, yield_date=yield_date, cycle_days=max(cycle_days, 0))
+
+    eligible_seconds = int((window_end - window_start).total_seconds())
+    online_seconds = min(_online_seconds(session, node, window_start, window_end, moment), eligible_seconds)
+    online_rate = ratio(online_seconds, eligible_seconds)
+
+    tasks = session.scalars(
+        select(ComputeTask).where(
+            ComputeTask.node_id == node.id,
+            ComputeTask.dispatched_at >= window_start,
+            ComputeTask.dispatched_at < window_end,
+        )
+    ).all()
+    tasks_dispatched = len(tasks)
+    tasks_completed = sum(1 for task in tasks if task.status == TaskStatus.COMPLETED)
+    response_rate = ratio(tasks_completed, tasks_dispatched) if tasks_dispatched else Decimal("0")
+
+    if tasks_dispatched == 0:
+        applied_online = Decimal("1")
+        applied_task = Decimal("0")
+    else:
+        applied_online = to_decimal(settings.online_weight)
+        applied_task = to_decimal(settings.task_weight)
+    performance = rate(applied_online * online_rate + applied_task * response_rate)
+    compute_output = money(to_decimal(node.purchase_amount) * performance)
+    points_delta = money(compute_output / Decimal(cycle_days))
+    points_delta = min(points_delta, _remaining_points(session, node))
+
+    return YieldComputation(
+        eligible=True,
+        yield_date=yield_date,
+        online_seconds=online_seconds,
+        eligible_seconds=eligible_seconds,
+        online_rate=online_rate,
+        tasks_dispatched=tasks_dispatched,
+        tasks_completed=tasks_completed,
+        task_response_rate=response_rate,
+        online_weight=money(applied_online),
+        task_weight=money(applied_task),
+        performance=performance,
+        compute_output=compute_output,
+        points_delta=points_delta,
+        cycle_days=cycle_days,
+    )
+
+
+def settle_yield_date(session: Session, yield_date: date, now: datetime | None = None) -> SettleSummary:
+    """清算指定 UTC 日。每个节点当天已有流水时跳过，保证脚本重复执行不会重复入账。"""
+    moment = now or utcnow()
+    if yield_date > moment.date():
+        raise AppError("不能清算未来日期")
+    expire_overdue_tasks(session, moment)
+    day_start, day_end = day_bounds(yield_date)
+    nodes = session.scalars(
+        select(ComputeNode)
+        .where(
+            ComputeNode.licensed_at < day_end,
+            ComputeNode.license_expires_at > day_start,
+        )
+        .order_by(ComputeNode.id)
+    ).all()
+    created = 0
+    skipped = 0
+    total = Decimal("0")
+    partial = yield_date == moment.date()
+    for node in nodes:
+        exists = session.scalar(
+            select(NodeYield.id).where(NodeYield.node_id == node.id, NodeYield.yield_date == yield_date)
+        )
+        if exists is not None:
+            skipped += 1
+            continue
+        computed = compute_node_yield(session, node, yield_date, moment)
+        if not computed.eligible:
+            continue
+        try:
+            with session.begin_nested():
+                session.add(_to_row(node, computed, partial))
+        except IntegrityError:
+            skipped += 1
+            continue
+        created += 1
+        total += computed.points_delta
+    return SettleSummary(
+        yield_date=yield_date,
+        created=created,
+        skipped=skipped,
+        total_points=money(total),
+        is_partial=partial,
+    )
+
+
+def list_node_yields(session: Session, sn: str) -> list[NodeYield]:
+    node = get_node_by_sn(session, sn.strip())
+    return list(
+        session.scalars(
+            select(NodeYield).where(NodeYield.node_id == node.id).order_by(NodeYield.yield_date.desc())
+        ).all()
+    )
+
+
+def _online_seconds(
+    session: Session,
+    node: ComputeNode,
+    window_start: datetime,
+    window_end: datetime,
+    now: datetime,
+) -> int:
+    intervals = session.scalars(
+        select(UptimeInterval).where(
+            UptimeInterval.node_id == node.id,
+            UptimeInterval.started_at < window_end,
+            or_(UptimeInterval.ended_at.is_(None), UptimeInterval.ended_at > window_start),
+        )
+    ).all()
+    total = 0
+    for interval in intervals:
+        start = max(interval.started_at, window_start)
+        end = min(_proven_end(interval, node, now), window_end)
+        if end > start:
+            total += int((end - start).total_seconds())
+    return total
+
+
+def _proven_end(interval: UptimeInterval, node: ComputeNode, now: datetime) -> datetime:
+    """开放区间只能证明到最后一次心跳；心跳仍在超时窗口内才延续到当前时间。"""
+    if interval.ended_at is not None:
+        return interval.ended_at
+    last_seen = node.last_heartbeat_at or interval.started_at
+    if last_seen + timedelta(minutes=settings.heartbeat_timeout_minutes) < now:
+        return last_seen
+    return now
+
+
+def _remaining_points(session: Session, node: ComputeNode) -> Decimal:
+    paid_rows = session.scalars(select(NodeYield.points_delta).where(NodeYield.node_id == node.id)).all()
+    paid = sum((to_decimal(item) for item in paid_rows), Decimal("0"))
+    remaining = to_decimal(node.purchase_amount) - paid
+    if remaining <= 0:
+        return Decimal("0")
+    return money(remaining)
+
+
+def _to_row(node: ComputeNode, computed: YieldComputation, partial: bool) -> NodeYield:
+    return NodeYield(
+        node_id=node.id,
+        user_id=node.user_id,
+        yield_date=computed.yield_date,
+        compute_output=computed.compute_output,
+        points_delta=computed.points_delta,
+        online_seconds=computed.online_seconds,
+        eligible_seconds=computed.eligible_seconds,
+        online_rate=computed.online_rate,
+        tasks_dispatched=computed.tasks_dispatched,
+        tasks_completed=computed.tasks_completed,
+        task_response_rate=computed.task_response_rate,
+        online_weight=computed.online_weight,
+        task_weight=computed.task_weight,
+        performance=computed.performance,
+        cycle_days=computed.cycle_days,
+        status=YieldStatus.SETTLED,
+        is_partial=partial,
+        created_at=utcnow(),
+    )