Add a node with one pasted command: sudo mode, stdin token, safe uninstall

CONTRIBUTING.md

@@ -822,7 +822,9 @@ installation, generation, specification digest and release before writing node f
 registering or reconnecting. Reject drift rather than overwriting retained identity
 or using local resource defaults. Registration consumes a token only after these
 checks. The installer verifies downloaded files and starts the ordinary node process
-as a user service; it performs no SSH installation, Session creation or model call.
+as a user service, or, run as root, as a root-owned system service for the dedicated
+`parsar-node` user it prepares; it performs no SSH installation, Session creation or
+model call.
 
 The [Hosted Sandbox Manager](services/agents-api/HOSTED-SANDBOX-MANAGER.md) is a
 deployment-level admin surface, separate from Project credentials. The paired
@@ -1856,11 +1858,29 @@ Docker socket or node identity mount in either mode. The ordinary standalone nod
 service owns its provider processes outside the Core container. Its `KillMode=process`
 preserves resident microVM/helper processes across a node-service restart. User KVM
 access and the Linux runtime libraries are prerequisites for microsandbox. The node
-runs as a systemd user service and needs linger. The only other launcher allowed
-is the root-run node installation (sudo mode) planned for phase 2b: one root-owned
-system service per installation that runs the same node program as a dedicated
-unprivileged service user. Do not add any other launcher, scheduler or recovery
-path. Node services restart after failures without a start limit, so a node
+installed by a normal user runs as a systemd user service and needs linger. Run as
+root (sudo mode), the installer instead prepares the host: it creates or adopts the
+`parsar-node` system user, adds it to the `docker` or `kvm` device group (no other
+group), and installs one root-owned system service per installation that runs the
+same node program with `User=parsar-node`. Sudo mode serves one Core per host,
+because its nodes share that account. Docker group membership makes that user,
+and so the node, root-equivalent on the host; that is inherent to Docker sandboxes,
+not a least-privilege boundary. Microsandbox needs only `kvm`. Files the service
+user owns are read, written and deleted only with its credentials, never by root,
+in a child that starts its own session with /dev/null as input, so nothing it runs
+can reach the administrator's terminal. That child also joins a new session
+keyring and dies with its parent, and root shows its output only as plain text
+(terminal controls become `?`). The installer turns SIGINT, SIGHUP and SIGTERM
+into stopping that child and what it started, which would otherwise outlive a
+closed terminal. Root never runs a file that user can write,
+opens a URL it wrote, or follows a link in its home. Sudo mode
+never installs Docker, KVM or packages, never changes device permissions, refuses
+SELinux-enforcing hosts and a token in the environment, and changes nothing when a
+check fails. `--uninstall` removes a node only after Core rejects its credential,
+never touches sandboxes, volumes or images (it keeps the Runtime image and the
+microsandbox store), deletes the account only when the installer created it and no
+node remains, and otherwise removes only the groups it added. Do not add any other launcher, scheduler or
+recovery path. Node services restart after failures without a start limit, so a node
 outlasts a Core outage, and stop restarting when the node program exits 78
 because Core answered 401 to its credential (a removed or retired node).
 The basic API image and binary builds remain independent artifacts.
@@ -1872,7 +1892,8 @@ microsandbox selections use Web's Standard size from
 `apps/web/src/features/sandbox/standard-sizes.json`, which the distribution build
 copies into the bundle; keep no second copy of those values. An existing database
 selection is never overwritten by installer defaults. Node configuration and identity live under
-`~/.parsar/nodes/<installation-id>/`; microsandbox uses its separate short private
+`~/.parsar/nodes/<installation-id>/` in the node account's home (`/var/lib/parsar-node`
+in sudo mode); microsandbox uses its separate short private
 Runtime home. Zero-node installs create no node identity state but retain the paired
 Core key for first setup.
 

deploy/install/distribution.py

@@ -147,44 +147,138 @@ def obtain_artifact(manifest, logical_path, destination, offline_root=None):
             if candidate.is_symlink() or not candidate.is_file():
                 raise DistributionError('Offline artifact must be a regular file: ' + logical_path)
             source = candidate
+    if source is not None:
+        return copy_artifact(source, target, entry, logical_path)
     base = manifest.get('artifact_base_url', '')
-    if source is None:
-        if not isinstance(base, str) or not base or urlsplit(base).query:
-            raise DistributionError('No downloadable artifact source; use the matching offline bundle')
-        url = safe_url(base.rstrip('/') + '/' + entry['filename'])
+    if not isinstance(base, str) or not base or urlsplit(base).query:
+        raise DistributionError('No downloadable artifact source; use the matching offline bundle')
+    url = safe_url(base.rstrip('/') + '/' + entry['filename'])
+    # A private partial file survives interruptions and reruns; the next attempt asks
+    # for the missing bytes only. The complete file is still verified as a whole.
+    partial = target.with_name('.' + target.name + '.partial')
     for attempt in range(3):
-        fd, temporary = tempfile.mkstemp(prefix='.artifact-', dir=target.parent)
         try:
-            with os.fdopen(fd, 'wb') as output:
-                stream = source.open('rb') if source else urllib.request.build_opener(NoRedirect()).open(url, timeout=30)
-                with stream:
-                    count = 0
-                    while True:
-                        block = stream.read(1024 * 1024)
-                        if not block:
-                            break
-                        count += len(block)
-                        if count > entry['size']:
-                            raise DistributionError('Artifact exceeds published size: ' + logical_path)
-                        output.write(block)
-                    if count != entry['size']:
-                        raise http.client.IncompleteRead(b'', entry['size'] - count)
-            if digest(temporary) != entry['sha256']:
-                raise DistributionError('Artifact checksum mismatch: ' + logical_path)
-            os.chmod(temporary, 0o700 if logical_path.startswith('native/') else 0o600)
-            os.replace(temporary, target)
-            return target
+            download_partial(url, partial, entry, logical_path)
+            break
         except urllib.error.HTTPError as error:
-            if error.code not in (408, 429, 500, 502, 503, 504) or attempt == 2:
-                raise DistributionError(f'Artifact download failed (HTTP {error.code}): {logical_path}. Check release access and retry.') from None
+            if error.code == 416:
+                discard_partial(partial)
+            elif error.code not in (408, 429, 500, 502, 503, 504) or attempt == 2:
+                raise DistributionError(f'Artifact download failed (HTTP {error.code}): {logical_path}. Check the console and retry.') from None
         except (urllib.error.URLError, socket.timeout, ConnectionError, http.client.HTTPException):
-            if attempt == 2 or source:
-                raise DistributionError('Artifact transfer interrupted: ' + logical_path + '. Check network access and rerun; installed state is retained.') from None
-        finally:
-            if os.path.exists(temporary):
-                os.unlink(temporary)
+            if attempt == 2:
+                raise DistributionError('Artifact transfer interrupted: ' + logical_path + '. Check network access and rerun; '
+                                        'the download resumes where it stopped.') from None
         time.sleep(attempt + 1)
-    raise DistributionError('Artifact download did not complete')
+    else:
+        raise DistributionError('Artifact download did not complete')
+    if digest(partial) != entry['sha256']:
+        discard_partial(partial)
+        raise DistributionError('Artifact checksum mismatch: ' + logical_path)
+    os.chmod(partial, 0o700 if logical_path.startswith('native/') else 0o600)
+    os.replace(partial, target)
+    validator_path(partial).unlink(missing_ok=True)
+    return target
+
+
+def validator_path(partial):
+    return partial.with_name(partial.name + '.validator')
+
+
+def discard_partial(partial):
+    partial.unlink(missing_ok=True)
+    validator_path(partial).unlink(missing_ok=True)
+
+
+def copy_artifact(source, target, entry, logical_path):
+    fd, temporary = tempfile.mkstemp(prefix='.artifact-', dir=target.parent)
+    try:
+        with os.fdopen(fd, 'wb') as output, source.open('rb') as stream:
+            count = 0
+            for block in iter(lambda: stream.read(1024 * 1024), b''):
+                count += len(block)
+                if count > entry['size']:
+                    raise DistributionError('Artifact exceeds published size: ' + logical_path)
+                output.write(block)
+        if count != entry['size'] or digest(temporary) != entry['sha256']:
+            raise DistributionError('Artifact checksum mismatch: ' + logical_path)
+        os.chmod(temporary, 0o700 if logical_path.startswith('native/') else 0o600)
+        os.replace(temporary, target)
+        return target
+    finally:
+        if os.path.exists(temporary):
+            os.unlink(temporary)
+
+
+# A download that brings fewer bytes than this in a window stops; a rerun resumes it.
+SLOW_SECONDS, SLOW_BYTES = 60, 64 * 1024
+
+
+def download_partial(url, partial, entry, logical_path):
+    """Complete the partial file, asking only for the bytes it is missing."""
+    size = entry['size']
+    offset = 0
+    if partial.is_symlink() or (partial.exists() and not partial.is_file()):
+        raise DistributionError('Partial download must be a regular file: ' + logical_path)
+    if partial.exists():
+        info = partial.stat()
+        if info.st_uid != os.getuid():
+            raise DistributionError('Partial download must be owned by this user: ' + logical_path)
+        offset = info.st_size if info.st_size <= size else 0
+    if offset == size:
+        return
+    headers = {}
+    if offset:
+        headers['Range'] = f'bytes={offset}-'
+        # If-Range makes a server whose file changed since the partial began send it whole.
+        try:
+            validator = validator_path(partial).read_text().strip()
+        except OSError:
+            validator = ''
+        if validator:
+            headers['If-Range'] = validator
+    request = urllib.request.Request(url, headers=headers)
+    with urllib.request.build_opener(NoRedirect()).open(request, timeout=30) as stream:
+        if offset and (stream.status != 206 or not stream.headers.get('Content-Range', '').startswith(f'bytes {offset}-')):
+            offset = 0  # The server sent the whole file; start over.
+        if not offset:
+            validator = stream.headers.get('ETag') or stream.headers.get('Last-Modified') or ''
+            if validator and all(32 <= ord(c) < 127 for c in validator) and len(validator) < 200:
+                with os.fdopen(os.open(validator_path(partial), os.O_WRONLY | os.O_CREAT | os.O_TRUNC | os.O_NOFOLLOW, 0o600), 'w') as note:
+                    note.write(validator)
+            else:
+                validator_path(partial).unlink(missing_ok=True)
+        flags = os.O_WRONLY | os.O_CREAT | os.O_NOFOLLOW | (os.O_APPEND if offset else os.O_TRUNC)
+        with os.fdopen(os.open(partial, flags, 0o600), 'ab' if offset else 'wb') as output:
+            count, reported = offset, offset * 10 // size
+            window, window_count = time.monotonic(), 0
+            while True:
+                block = stream.read1(1024 * 1024)
+                if not block:
+                    break
+                count += len(block)
+                if count > size:
+                    output.close()
+                    discard_partial(partial)
+                    raise DistributionError('Artifact exceeds published size: ' + logical_path)
+                output.write(block)
+                window_count += len(block)
+                if time.monotonic() - window >= SLOW_SECONDS:
+                    if window_count < SLOW_BYTES:
+                        raise DistributionError(f'Artifact download stalled (under {SLOW_BYTES // 1024} KiB in {SLOW_SECONDS} s): '
+                                                f'{logical_path}. The downloaded part is kept; check the network, then rerun '
+                                                'the command to resume.')
+                    window, window_count = time.monotonic(), 0
+                if size >= 50 * 1024 * 1024 and count * 10 // size > reported:
+                    reported = count * 10 // size
+                    try:
+                        print(f'Downloading {logical_path}: {reported * 10}% of {size // (1024 * 1024)} MiB', flush=True)
+                    except BrokenPipeError:
+                        # Nobody reads the output any more (the installer's terminal is gone);
+                        # stop instead of retrying it as a network error.
+                        raise DistributionError('Output closed; stopped downloading ' + logical_path) from None
+        if count != size:
+            raise http.client.IncompleteRead(b'', size - count)
 
 
 def runtime_archive(manifest, cache_root, offline_root=None):