diff --git a/src-tauri/src/core/remote_bootstrap.rs b/src-tauri/src/core/remote_bootstrap.rs
index 69d46f37..60c1fcdc 100644
--- a/src-tauri/src/core/remote_bootstrap.rs
+++ b/src-tauri/src/core/remote_bootstrap.rs
@@ -140,10 +140,112 @@ echo "treq bootstrap: manifest version $TREQ_MANIFEST_VERSION installed"
   )
 }
 
+/// Idempotent shell command that installs the Treq SSH CA's public key on a
+/// managed VM and points `sshd`'s `TrustedUserCAKeys` at it (PRD "Configure
+/// the managed VM to trust the Treq SSH CA"). Mirrors
+/// `_shared/remote/ssh-vm-config.ts::installCaTrustCommand` in the Edge
+/// Function, which is the path actually wired into provisioning today; this
+/// Rust copy exists so a native (Phase 4) transport can push the same
+/// config without going through the control plane's exec channel.
+pub fn install_ca_trust_command(ca_public_key_line: &str) -> Vec<String> {
+  let script = format!(
+    r#"#!/bin/sh
+set -eu
+mkdir -p /etc/ssh/sshd_config.d
+cat > /etc/ssh/treq_ca.pub <<'TREQ_CA_EOF'
+{ca_public_key_line}
+TREQ_CA_EOF
+cat > /etc/ssh/sshd_config.d/60-treq-ca.conf <<'TREQ_SSHD_EOF'
+TrustedUserCAKeys /etc/ssh/treq_ca.pub
+TREQ_SSHD_EOF
+if command -v systemctl >/dev/null 2>&1 && systemctl is-active sshd >/dev/null 2>&1; then
+  systemctl reload sshd
+elif command -v systemctl >/dev/null 2>&1 && systemctl is-active ssh >/dev/null 2>&1; then
+  systemctl reload ssh
+elif [ -f /var/run/sshd.pid ]; then
+  kill -HUP "$(cat /var/run/sshd.pid)"
+fi
+echo "treq: CA trust installed"
+"#
+  );
+  vec!["/bin/sh".to_string(), "-c".to_string(), script]
+}
+
+/// Idempotent authorized_keys install for the direct existing-key auth
+/// alternative (PRD "Existing keys without certificates"). Guards on a
+/// fingerprint marker comment so a repeated install never duplicates the
+/// entry. Mirrors `_shared/remote/ssh-vm-config.ts::installAuthorizedKeyCommand`.
+pub fn install_authorized_key_command(
+  public_key_line: &str,
+  fingerprint_sha256: &str,
+) -> Vec<String> {
+  let marker = format!("# treq-client-key:{fingerprint_sha256}");
+  let script = format!(
+    r#"#!/bin/sh
+set -eu
+mkdir -p /home/treq/.ssh
+chmod 700 /home/treq/.ssh
+touch /home/treq/.ssh/authorized_keys
+if ! grep -qF "{marker}" /home/treq/.ssh/authorized_keys 2>/dev/null; then
+  printf '%s\n%s\n' "{marker}" "{public_key_line}" >> /home/treq/.ssh/authorized_keys
+fi
+chmod 600 /home/treq/.ssh/authorized_keys
+echo "treq: authorized key installed"
+"#
+  );
+  vec!["/bin/sh".to_string(), "-c".to_string(), script]
+}
+
+/// Removes exactly the marker + key line pair `install_authorized_key_command`
+/// added for `fingerprint_sha256`, leaving every other entry untouched.
+/// Mirrors `_shared/remote/ssh-vm-config.ts::removeAuthorizedKeyCommand`.
+pub fn remove_authorized_key_command(fingerprint_sha256: &str) -> Vec<String> {
+  let marker = format!("# treq-client-key:{fingerprint_sha256}");
+  let script = format!(
+    r#"#!/bin/sh
+set -eu
+FILE=/home/treq/.ssh/authorized_keys
+if [ -f "$FILE" ]; then
+  MARKER="{marker}"
+  awk -v marker="$MARKER" '
+    $0 == marker {{ skip = 2; next }}
+    skip > 0 {{ skip--; next }}
+    {{ print }}
+  ' "$FILE" > "$FILE.treq_tmp"
+  mv "$FILE.treq_tmp" "$FILE"
+fi
+echo "treq: authorized key removed"
+"#
+  );
+  vec!["/bin/sh".to_string(), "-c".to_string(), script]
+}
+
 #[cfg(test)]
 mod tests {
   use super::*;
 
+  #[test]
+  fn ca_trust_command_installs_trusted_user_ca_keys() {
+    let command = install_ca_trust_command("ssh-ed25519 AAAA... treq-ssh-ca");
+    assert_eq!(command[0], "/bin/sh");
+    assert!(command[2].contains("ssh-ed25519 AAAA... treq-ssh-ca"));
+    assert!(command[2].contains("TrustedUserCAKeys /etc/ssh/treq_ca.pub"));
+  }
+
+  #[test]
+  fn authorized_key_install_is_guarded_by_fingerprint_marker() {
+    let command = install_authorized_key_command("ssh-ed25519 AAAA... me", "SHA256:abc");
+    assert!(command[2].contains("# treq-client-key:SHA256:abc"));
+    assert!(command[2].contains("grep -qF"));
+  }
+
+  #[test]
+  fn authorized_key_removal_targets_the_same_marker() {
+    let command = remove_authorized_key_command("SHA256:abc");
+    assert!(command[2].contains("# treq-client-key:SHA256:abc"));
+    assert!(command[2].contains("skip = 2"));
+  }
+
   #[test]
   fn known_manifest_version_resolves() {
     assert!(manifest_for_version(1).is_some());
diff --git a/src-tauri/src/core/remote_control_plane.rs b/src-tauri/src/core/remote_control_plane.rs
index c20a345d..84d99ad3 100644
--- a/src-tauri/src/core/remote_control_plane.rs
+++ b/src-tauri/src/core/remote_control_plane.rs
@@ -154,6 +154,54 @@ pub struct IssueCertificateResponse {
   pub endpoint: SshEndpoint,
 }
 
+/// Direct existing-key auth alternative (PRD "Existing keys without
+/// certificates"): installs or removes a registered public key from the
+/// managed VM's `authorized_keys`. Both directions are idempotent and
+/// auditable on the server side.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct InstallAuthorizedKeyRequest {
+  pub instance_id: String,
+  pub key_id: String,
+  pub idempotency_key: IdempotencyKey,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct RemoveAuthorizedKeyRequest {
+  pub instance_id: String,
+  pub key_id: String,
+  pub idempotency_key: IdempotencyKey,
+}
+
+/// A recorded host-key rotation, per the PRD's "Reprovisioning may rotate
+/// the host key" paragraph: old and new fingerprints, the generation the new
+/// key was observed at, the provider resource it was scanned from, and who
+/// initiated the transition that produced it.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct HostKeyRotationRecord {
+  pub endpoint_id: String,
+  pub previous_fingerprint_sha256: Option<String>,
+  pub new_fingerprint_sha256: String,
+  pub generation: u64,
+  pub provider_resource_id: Option<String>,
+  pub initiating_principal: String,
+  pub rotated_at: String,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct ClientKeyRecord {
+  pub id: String,
+  pub algorithm: String,
+  pub fingerprint_sha256: String,
+  pub comment: Option<String>,
+  pub created_at: String,
+  pub revoked_at: Option<String>,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct ListClientKeysResponse {
+  pub keys: Vec<ClientKeyRecord>,
+}
+
 // -- User-managed endpoints ---------------------------------------------------
 
 /// Registers a fully explicit user-owned VM endpoint. Every field is supplied
diff --git a/supabase/functions/_shared/remote/audit.ts b/supabase/functions/_shared/remote/audit.ts
index 7342695c..e3efe366 100644
--- a/supabase/functions/_shared/remote/audit.ts
+++ b/supabase/functions/_shared/remote/audit.ts
@@ -21,7 +21,16 @@ export type RemoteAuditEventType =
   | "instance_delete_failed"
   | "host_key_registered"
   | "host_key_rotated"
-  | "readiness_stage_failed";
+  | "host_keyscan_failed"
+  | "readiness_stage_failed"
+  | "client_key_registered"
+  | "client_key_revoked"
+  | "certificate_issued"
+  | "certificate_issue_failed"
+  | "authorized_key_installed"
+  | "authorized_key_removed"
+  | "ca_trust_installed"
+  | "ca_trust_install_failed";
 
 export async function recordAuditEvent(
   supabase: SupabaseClient,
diff --git a/supabase/functions/_shared/remote/client-key-store.ts b/supabase/functions/_shared/remote/client-key-store.ts
new file mode 100644
index 00000000..ccd9007f
--- /dev/null
+++ b/supabase/functions/_shared/remote/client-key-store.ts
@@ -0,0 +1,140 @@
+// Storage helpers for remote_client_keys and remote_endpoint_authorized_keys,
+// used by the remote-ssh-trust Edge Function. Only public key material and
+// metadata are ever written here (PRD "Client key policy").
+
+import type { SupabaseClient } from "@supabase/supabase-js";
+
+export interface ClientKeyRow {
+  id: string;
+  owner_user_id: string;
+  public_key: string;
+  algorithm: string;
+  fingerprint_sha256: string;
+  comment: string | null;
+  created_at: string;
+  revoked_at: string | null;
+}
+
+export async function findClientKeyByFingerprint(
+  supabase: SupabaseClient,
+  ownerUserId: string,
+  fingerprintSha256: string,
+): Promise<ClientKeyRow | null> {
+  const { data, error } = await supabase
+    .from("remote_client_keys")
+    .select("*")
+    .eq("owner_user_id", ownerUserId)
+    .eq("fingerprint_sha256", fingerprintSha256)
+    .maybeSingle();
+  if (error) throw new Error(`failed to look up client key: ${error.message}`);
+  return data as ClientKeyRow | null;
+}
+
+export async function insertClientKey(
+  supabase: SupabaseClient,
+  params: { ownerUserId: string; publicKey: string; algorithm: string; fingerprintSha256: string; comment: string | null },
+): Promise<ClientKeyRow> {
+  const { data, error } = await supabase
+    .from("remote_client_keys")
+    .insert({
+      owner_user_id: params.ownerUserId,
+      public_key: params.publicKey,
+      algorithm: params.algorithm,
+      fingerprint_sha256: params.fingerprintSha256,
+      comment: params.comment,
+    })
+    .select()
+    .single();
+  if (error) throw new Error(`failed to register client key: ${error.message}`);
+  return data as ClientKeyRow;
+}
+
+export async function listClientKeys(supabase: SupabaseClient, ownerUserId: string): Promise<ClientKeyRow[]> {
+  const { data, error } = await supabase
+    .from("remote_client_keys")
+    .select("*")
+    .eq("owner_user_id", ownerUserId)
+    .order("created_at", { ascending: false });
+  if (error) throw new Error(`failed to list client keys: ${error.message}`);
+  return (data ?? []) as ClientKeyRow[];
+}
+
+// Ownership is enforced by the `owner_user_id` filter here, not by trusting
+// the caller-supplied key id alone (PRD "Security requirements").
+export async function getOwnedClientKey(
+  supabase: SupabaseClient,
+  ownerUserId: string,
+  keyId: string,
+): Promise<ClientKeyRow | null> {
+  const { data, error } = await supabase
+    .from("remote_client_keys")
+    .select("*")
+    .eq("owner_user_id", ownerUserId)
+    .eq("id", keyId)
+    .maybeSingle();
+  if (error) throw new Error(`failed to read client key: ${error.message}`);
+  return data as ClientKeyRow | null;
+}
+
+export async function revokeClientKey(supabase: SupabaseClient, ownerUserId: string, keyId: string): Promise<void> {
+  const { error } = await supabase
+    .from("remote_client_keys")
+    .update({ revoked_at: new Date().toISOString() })
+    .eq("owner_user_id", ownerUserId)
+    .eq("id", keyId);
+  if (error) throw new Error(`failed to revoke client key: ${error.message}`);
+}
+
+export interface AuthorizedKeyRow {
+  id: string;
+  endpoint_id: string;
+  client_key_id: string;
+  installed_at: string;
+  removed_at: string | null;
+}
+
+export async function findActiveAuthorizedKey(
+  supabase: SupabaseClient,
+  endpointId: string,
+  clientKeyId: string,
+): Promise<AuthorizedKeyRow | null> {
+  const { data, error } = await supabase
+    .from("remote_endpoint_authorized_keys")
+    .select("*")
+    .eq("endpoint_id", endpointId)
+    .eq("client_key_id", clientKeyId)
+    .is("removed_at", null)
+    .maybeSingle();
+  if (error) throw new Error(`failed to look up authorized key record: ${error.message}`);
+  return data as AuthorizedKeyRow | null;
+}
+
+export async function recordAuthorizedKeyInstalled(
+  supabase: SupabaseClient,
+  params: { ownerUserId: string; endpointId: string; clientKeyId: string },
+): Promise<void> {
+  const { error } = await supabase.from("remote_endpoint_authorized_keys").upsert(
+    {
+      owner_user_id: params.ownerUserId,
+      endpoint_id: params.endpointId,
+      client_key_id: params.clientKeyId,
+      installed_at: new Date().toISOString(),
+      removed_at: null,
+    },
+    { onConflict: "endpoint_id,client_key_id" },
+  );
+  if (error) throw new Error(`failed to record authorized key install: ${error.message}`);
+}
+
+export async function recordAuthorizedKeyRemoved(
+  supabase: SupabaseClient,
+  endpointId: string,
+  clientKeyId: string,
+): Promise<void> {
+  const { error } = await supabase
+    .from("remote_endpoint_authorized_keys")
+    .update({ removed_at: new Date().toISOString() })
+    .eq("endpoint_id", endpointId)
+    .eq("client_key_id", clientKeyId);
+  if (error) throw new Error(`failed to record authorized key removal: ${error.message}`);
+}
diff --git a/supabase/functions/_shared/remote/instance-store.ts b/supabase/functions/_shared/remote/instance-store.ts
index 4f6243be..aa244cc9 100644
--- a/supabase/functions/_shared/remote/instance-store.ts
+++ b/supabase/functions/_shared/remote/instance-store.ts
@@ -10,7 +10,19 @@
 import type { SupabaseClient } from "@supabase/supabase-js";
 import type { RegionCode, SizePreset } from "./catalog.ts";
 
-export type OperationType = "provision" | "wake" | "reprovision" | "delete";
+export type OperationType =
+  | "provision"
+  | "wake"
+  | "reprovision"
+  | "delete"
+  | "register_client_key"
+  | "revoke_client_key"
+  | "issue_certificate"
+  | "register_endpoint"
+  | "register_repository"
+  | "install_authorized_key"
+  | "remove_authorized_key"
+  | "keyscan_host_key";
 export type OperationStatus = "pending" | "in_progress" | "succeeded" | "failed";
 
 export interface OperationRow {
diff --git a/supabase/functions/_shared/remote/sprites-adapter.ts b/supabase/functions/_shared/remote/sprites-adapter.ts
index 1830ae86..933d42b7 100644
--- a/supabase/functions/_shared/remote/sprites-adapter.ts
+++ b/supabase/functions/_shared/remote/sprites-adapter.ts
@@ -63,12 +63,21 @@ export interface ReplaceInstanceParams {
   idempotencyKey: string;
 }
 
+export interface MachineExecResult {
+  exitCode: number;
+  stdout: string;
+  stderr: string;
+}
+
 export interface ManagedComputeProvider {
   createInstance(params: CreateInstanceParams): Promise<ProviderInstance>;
   getInstance(providerId: string): Promise<ProviderInstance>;
   wakeInstance(providerId: string): Promise<void>;
   replaceInstance(params: ReplaceInstanceParams): Promise<ProviderInstance>;
   deleteInstance(providerId: string): Promise<void>;
+  // Runs a command inside a running machine (Fly Machines `/exec`), used to
+  // install CA trust and authorized_keys onto an already-booted managed VM.
+  execOnMachine(providerId: string, command: string[], timeoutSeconds?: number): Promise<MachineExecResult>;
 }
 
 function sizeToGuest(preset: SizePreset) {
@@ -159,6 +168,30 @@ export class SpritesProvider implements ManagedComputeProvider {
     return `${this.machinesUrl()}/${id}`;
   }
 
+  // Runs a command inside a running machine via the Fly Machines `/exec`
+  // endpoint. This is how server-side config (CA trust, authorized_keys) is
+  // pushed onto an already-booted managed VM without a native SSH client -
+  // the same mechanism `init.exec` uses at boot, just invoked after the fact.
+  async execOnMachine(providerId: string, command: string[], timeoutSeconds = 20): Promise<MachineExecResult> {
+    let response: Response;
+    try {
+      response = await fetch(`${this.machineUrl(providerId)}/exec`, {
+        method: "POST",
+        headers: this.headers(),
+        body: JSON.stringify({ cmd: command, timeout: timeoutSeconds }),
+      });
+    } catch (err) {
+      throw new ProviderError("unavailable", `could not reach Fly Machines API: ${(err as Error).message}`);
+    }
+    if (!response.ok) throw await this.mapErrorResponse(response);
+    const body = await response.json();
+    return {
+      exitCode: typeof body.exit_code === "number" ? body.exit_code : -1,
+      stdout: typeof body.stdout === "string" ? body.stdout : "",
+      stderr: typeof body.stderr === "string" ? body.stderr : "",
+    };
+  }
+
   private headers(idempotencyKey?: string): HeadersInit {
     const headers: Record<string, string> = {
       Authorization: `Bearer ${this.config.apiToken}`,
diff --git a/supabase/functions/_shared/remote/ssh-cert.ts b/supabase/functions/_shared/remote/ssh-cert.ts
new file mode 100644
index 00000000..2aeddd9d
--- /dev/null
+++ b/supabase/functions/_shared/remote/ssh-cert.ts
@@ -0,0 +1,209 @@
+// Real OpenSSH user certificate signing (`ssh-ed25519-cert-v01@openssh.com`),
+// per PRD "SSH identity and certificates". The CA private key never leaves
+// this module: it is read once from an Edge Function secret, used to sign,
+// and discarded - it is never returned in a response or written to a table.
+//
+// Wire format: OpenSSH's PROTOCOL.certkeys, "ssh-ed25519-cert-v01@openssh.com"
+// section:
+//
+//   string    "ssh-ed25519-cert-v01@openssh.com"
+//   string    nonce
+//   string    pk (the 32-byte ed25519 public key being certified)
+//   uint64    serial
+//   uint32    type (1 = user)
+//   string    key id
+//   string    valid principals
+//   uint64    valid after
+//   uint64    valid before
+//   string    critical options
+//   string    extensions
+//   string    reserved
+//   string    signature key (CA's own public key blob)
+//   string    signature
+//
+// Only ed25519 user keys are certified today. RSA/ECDSA support is a real
+// gap, not a stub: signing those cert types needs different key material
+// handling and is left for a follow-up rather than faked here.
+
+import { SshReader, SshWriter } from "./ssh-wire.ts";
+import { base64Decode, base64Encode } from "./ssh-wire.ts";
+
+const CERT_TYPE_ED25519 = "ssh-ed25519-cert-v01@openssh.com";
+const CA_KEY_TYPE_ED25519 = "ssh-ed25519";
+const SSH_CERT_TYPE_USER = 1;
+
+// Extensions granted to every issued certificate: the standard interactive
+// permissions OpenSSH clients expect for a login/shell/exec session. No
+// critical options are set.
+const DEFAULT_EXTENSIONS = [
+  "permit-X11-forwarding",
+  "permit-agent-forwarding",
+  "permit-port-forwarding",
+  "permit-pty",
+  "permit-user-rc",
+];
+
+export interface CaKeyMaterial {
+  /// Raw 32-byte ed25519 seed. Held only in memory for the duration of one
+  /// signing call.
+  privateSeed: Uint8Array;
+  /// Raw 32-byte ed25519 public key, matching `privateSeed`.
+  publicKey: Uint8Array;
+}
+
+// Reads the CA key pair from Edge Function secrets. Both are base64-encoded
+// raw 32-byte values (not OpenSSH-formatted lines) so this module owns the
+// one place that turns them into wire-format blobs.
+export function caKeyMaterialFromEnv(): CaKeyMaterial {
+  const seedB64 = Deno.env.get("REMOTE_SSH_CA_ED25519_SEED_BASE64");
+  const publicB64 = Deno.env.get("REMOTE_SSH_CA_ED25519_PUBLIC_KEY_BASE64");
+  if (!seedB64 || !publicB64) {
+    throw new Error(
+      "REMOTE_SSH_CA_ED25519_SEED_BASE64 and REMOTE_SSH_CA_ED25519_PUBLIC_KEY_BASE64 must be set as Edge Function secrets",
+    );
+  }
+  const privateSeed = base64Decode(seedB64);
+  const publicKey = base64Decode(publicB64);
+  if (privateSeed.length !== 32 || publicKey.length !== 32) {
+    throw new Error("SSH CA key material must be raw 32-byte ed25519 values");
+  }
+  return { privateSeed, publicKey };
+}
+
+// SSH-wire-format blob for an ssh-ed25519 public key: string "ssh-ed25519",
+// string <32 bytes>.
+export function ed25519PublicKeyBlob(rawPublicKey: Uint8Array): Uint8Array {
+  return new SshWriter().writeString(CA_KEY_TYPE_ED25519).writeString(rawPublicKey).toBytes();
+}
+
+// The OpenSSH `authorized_keys`/`TrustedUserCAKeys`-format public key line
+// for the CA, e.g. "ssh-ed25519 AAAA... treq-ssh-ca".
+export function caPublicKeyLine(ca: CaKeyMaterial, comment = "treq-ssh-ca"): string {
+  const blob = ed25519PublicKeyBlob(ca.publicKey);
+  return `${CA_KEY_TYPE_ED25519} ${base64Encode(blob)} ${comment}`;
+}
+
+// PKCS8 DER wrapper for a raw ed25519 seed (RFC 8410): a fixed 16-byte
+// prefix followed by the 32-byte seed, since ed25519 PKCS8 has no variable
+// fields at this key size.
+const PKCS8_ED25519_PREFIX = new Uint8Array([
+  0x30, 0x2e, 0x02, 0x01, 0x00, 0x30, 0x05, 0x06, 0x03, 0x2b, 0x65, 0x70, 0x04, 0x22, 0x04, 0x20,
+]);
+
+async function importEd25519PrivateKey(seed: Uint8Array): Promise<CryptoKey> {
+  const der = new Uint8Array(PKCS8_ED25519_PREFIX.length + seed.length);
+  der.set(PKCS8_ED25519_PREFIX, 0);
+  der.set(seed, PKCS8_ED25519_PREFIX.length);
+  return await crypto.subtle.importKey("pkcs8", der as BufferSource, { name: "Ed25519" }, false, ["sign"]);
+}
+
+async function signEd25519(seed: Uint8Array, message: Uint8Array): Promise<Uint8Array> {
+  const key = await importEd25519PrivateKey(seed);
+  const signature = await crypto.subtle.sign("Ed25519", key, message as BufferSource);
+  return new Uint8Array(signature);
+}
+
+export interface IssueCertificateParams {
+  ca: CaKeyMaterial;
+  /// Raw 32-byte ed25519 public key of the user's registered client key.
+  userPublicKey: Uint8Array;
+  principals: string[];
+  /// Unique per-certificate serial for audit correlation. Must fit in a
+  /// uint64; caller is responsible for uniqueness (a random 63-bit value is
+  /// used by `randomSerial`).
+  serial: bigint;
+  keyId: string;
+  validAfter: Date;
+  validBefore: Date;
+}
+
+export interface IssuedCertificate {
+  /// Full "ssh-ed25519-cert-v01@openssh.com <base64> <comment>" line, ready
+  /// to write to a certificate file next to the user's private key.
+  certificateLine: string;
+  serial: string;
+}
+
+export function randomSerial(): bigint {
+  const bytes = crypto.getRandomValues(new Uint8Array(8));
+  bytes[0] &= 0x7f; // keep it a positive int64 for readability in logs
+  let value = 0n;
+  for (const byte of bytes) value = (value << 8n) | BigInt(byte);
+  return value;
+}
+
+export async function issueEd25519UserCertificate(params: IssueCertificateParams): Promise<IssuedCertificate> {
+  if (params.userPublicKey.length !== 32) {
+    throw new Error("user public key must be a raw 32-byte ed25519 key");
+  }
+  const nonce = crypto.getRandomValues(new Uint8Array(32));
+
+  const toBeSigned = new SshWriter()
+    .writeString(CERT_TYPE_ED25519)
+    .writeString(nonce)
+    .writeString(params.userPublicKey)
+    .writeUint64(params.serial)
+    .writeUint32(SSH_CERT_TYPE_USER)
+    .writeString(params.keyId)
+    .writeNameList(params.principals)
+    .writeUint64(BigInt(Math.floor(params.validAfter.getTime() / 1000)))
+    .writeUint64(BigInt(Math.floor(params.validBefore.getTime() / 1000)))
+    .writeString("") // critical options: none
+    .writeString(encodeExtensions(DEFAULT_EXTENSIONS))
+    .writeString("") // reserved
+    .writeString(ed25519PublicKeyBlob(params.ca.publicKey))
+    .toBytes();
+
+  const signatureBytes = await signEd25519(params.ca.privateSeed, toBeSigned);
+  const signatureBlob = new SshWriter().writeString(CA_KEY_TYPE_ED25519).writeString(signatureBytes).toBytes();
+
+  const full = new SshWriter().writeRaw(toBeSigned).writeString(signatureBlob).toBytes();
+  const comment = params.keyId;
+  return {
+    certificateLine: `${CERT_TYPE_ED25519} ${base64Encode(full)} ${comment}`,
+    serial: params.serial.toString(),
+  };
+}
+
+// Extensions are encoded as a name/value map where every value here is
+// empty, per PROTOCOL.certkeys: each entry is `string name, string value`
+// concatenated and wrapped in one outer string.
+function encodeExtensions(names: string[]): Uint8Array {
+  const writer = new SshWriter();
+  for (const name of names) {
+    writer.writeString(name);
+    writer.writeString("");
+  }
+  return writer.toBytes();
+}
+
+// Parses back a certificate this module issued, for tests: returns the
+// fields relevant to verifying the signing round-trip.
+export function decodeCertificateForTests(certificateLine: string): {
+  serial: bigint;
+  principals: string[];
+  validAfter: bigint;
+  validBefore: bigint;
+} {
+  const [, encoded] = certificateLine.split(" ");
+  const bytes = base64Decode(encoded);
+  const reader = new SshReader(bytes);
+  reader.readUtf8String(); // cert type
+  reader.readString(); // nonce
+  reader.readString(); // pk
+  const serialBytes = reader.readBytes(8);
+  const serial = bytesToBigUint64(serialBytes);
+  reader.readUint32(); // type
+  reader.readString(); // key id
+  const principalsRaw = new TextDecoder().decode(reader.readString());
+  const principals = principalsRaw.length > 0 ? principalsRaw.split("\0") : [];
+  const validAfter = bytesToBigUint64(reader.readBytes(8));
+  const validBefore = bytesToBigUint64(reader.readBytes(8));
+  return { serial, principals, validAfter, validBefore };
+}
+
+function bytesToBigUint64(bytes: Uint8Array): bigint {
+  let value = 0n;
+  for (const byte of bytes) value = (value << 8n) | BigInt(byte);
+  return value;
+}
diff --git a/supabase/functions/_shared/remote/ssh-keys.ts b/supabase/functions/_shared/remote/ssh-keys.ts
new file mode 100644
index 00000000..f5296039
--- /dev/null
+++ b/supabase/functions/_shared/remote/ssh-keys.ts
@@ -0,0 +1,68 @@
+// OpenSSH public-key line parsing and fingerprinting, per the PRD's "Client
+// key policy": Treq only ever stores and reasons about public material.
+
+import { base64Decode, sshFingerprintSha256, SshReader } from "./ssh-wire.ts";
+
+export type SupportedKeyAlgorithm = "ssh-ed25519";
+
+export interface ParsedPublicKey {
+  algorithm: string;
+  /// Raw SSH-wire key blob (the full `string algorithm, ... key fields`
+  /// structure), exactly as it appears base64-encoded in an
+  /// authorized_keys-format line.
+  blob: Uint8Array;
+  comment: string | null;
+  fingerprintSha256: string;
+}
+
+export class UnsupportedKeyError extends Error {}
+
+// Parses a single "authorized_keys"-format line: "<algorithm> <base64> [comment]".
+export async function parseOpenSshPublicKey(line: string): Promise<ParsedPublicKey> {
+  const trimmed = line.trim();
+  const parts = trimmed.split(/\s+/);
+  if (parts.length < 2) {
+    throw new UnsupportedKeyError("public key must be in 'algorithm base64 [comment]' format");
+  }
+  const [algorithm, encoded, ...commentParts] = parts;
+  let blob: Uint8Array;
+  try {
+    blob = base64Decode(encoded);
+  } catch {
+    throw new UnsupportedKeyError("public key base64 payload is invalid");
+  }
+  // The blob's own embedded algorithm name must match the line's prefix -
+  // this is what stops a client from mislabeling a key's algorithm.
+  const reader = new SshReader(blob);
+  let embeddedAlgorithm: string;
+  try {
+    embeddedAlgorithm = reader.readUtf8String();
+  } catch {
+    throw new UnsupportedKeyError("public key blob is malformed");
+  }
+  if (embeddedAlgorithm !== algorithm) {
+    throw new UnsupportedKeyError("public key algorithm does not match its encoded blob");
+  }
+  const fingerprintSha256 = await sshFingerprintSha256(blob);
+  return {
+    algorithm,
+    blob,
+    comment: commentParts.length > 0 ? commentParts.join(" ") : null,
+    fingerprintSha256,
+  };
+}
+
+// Extracts the raw 32-byte ed25519 public key from a parsed ssh-ed25519 key
+// blob (string "ssh-ed25519", string <32-byte key>).
+export function extractEd25519RawKey(parsed: ParsedPublicKey): Uint8Array {
+  if (parsed.algorithm !== "ssh-ed25519") {
+    throw new UnsupportedKeyError(`unsupported key algorithm '${parsed.algorithm}'`);
+  }
+  const reader = new SshReader(parsed.blob);
+  reader.readUtf8String(); // algorithm name, already validated
+  const key = reader.readString();
+  if (key.length !== 32) {
+    throw new UnsupportedKeyError("ssh-ed25519 key blob has an unexpected length");
+  }
+  return key;
+}
diff --git a/supabase/functions/_shared/remote/ssh-keyscan.ts b/supabase/functions/_shared/remote/ssh-keyscan.ts
new file mode 100644
index 00000000..235629a2
--- /dev/null
+++ b/supabase/functions/_shared/remote/ssh-keyscan.ts
@@ -0,0 +1,209 @@
+// Real TCP-level SSH host-key scan: connects to an endpoint's SSH port,
+// performs the identification-string exchange and enough of the SSH
+// transport key-exchange (RFC 4253) to read the server's actual host public
+// key out of the wire, then disconnects. This is not a full SSH client - no
+// session is authenticated and no shared secret is used - it exists only to
+// populate `remote_endpoint_host_keys` with a real observed fingerprint
+// instead of a placeholder (see PRD "Host-key verification").
+//
+// Scope and honest limitations:
+//   - Only "curve25519-sha256" key exchange and "ssh-ed25519" host keys are
+//     requested. A server offering only RSA/ECDSA host keys is reported as
+//     an explicit `unsupported_host_key_algorithm` error rather than a
+//     fabricated fingerprint.
+//   - The client's ECDH ephemeral value is 32 random bytes, not a real X25519
+//     key pair, because reading the host key out of KEX_ECDH_REPLY does not
+//     require completing the exchange or deriving a shared secret. That also
+//     means this scan does NOT verify the server's exchange-hash signature,
+//     so it establishes trust on first observation rather than proving the
+//     TCP path was not tampered with. That is the inherent trust-on-first-use
+//     property of any first keyscan (matching plain `ssh-keyscan`'s own
+//     behavior) - later connections must pin against what this call records.
+
+import { parseOpenSshPublicKey, type ParsedPublicKey } from "./ssh-keys.ts";
+import { SshReader, SshWriter } from "./ssh-wire.ts";
+import { base64Encode } from "./ssh-wire.ts";
+
+const SSH_MSG_KEXINIT = 20;
+const SSH_MSG_KEX_ECDH_INIT = 30;
+const SSH_MSG_KEX_ECDH_REPLY = 31;
+
+export class KeyscanError extends Error {
+  constructor(
+    message: string,
+    public readonly kind:
+      | "connect_failed"
+      | "protocol_error"
+      | "unsupported_host_key_algorithm"
+      | "timeout",
+  ) {
+    super(message);
+  }
+}
+
+export interface ScannedHostKey {
+  algorithm: string;
+  fingerprintSha256: string;
+  /// authorized_keys-format line for the observed key ("<algo> <base64>"),
+  /// stored for display/debugging - never used as a client credential.
+  publicKeyLine: string;
+}
+
+async function readExactly(conn: Deno.Conn, length: number, deadline: number): Promise<Uint8Array> {
+  const out = new Uint8Array(length);
+  let filled = 0;
+  while (filled < length) {
+    if (Date.now() > deadline) throw new KeyscanError("timed out reading from SSH endpoint", "timeout");
+    const n = await conn.read(out.subarray(filled));
+    if (n === null) throw new KeyscanError("connection closed before host key was received", "protocol_error");
+    filled += n;
+  }
+  return out;
+}
+
+async function readLine(conn: Deno.Conn, deadline: number): Promise<string> {
+  const bytes: number[] = [];
+  const one = new Uint8Array(1);
+  while (bytes.length < 255) {
+    if (Date.now() > deadline) throw new KeyscanError("timed out reading SSH identification string", "timeout");
+    const n = await conn.read(one);
+    if (n === null) throw new KeyscanError("connection closed during identification exchange", "protocol_error");
+    if (one[0] === 0x0a) break;
+    if (one[0] !== 0x0d) bytes.push(one[0]);
+  }
+  return new TextDecoder().decode(new Uint8Array(bytes));
+}
+
+function framePacket(payload: Uint8Array): Uint8Array {
+  // packet_length(4) + padding_length(1) + payload + padding, padded to a
+  // multiple of 8 with at least 4 bytes of padding (RFC 4253 section 6).
+  let paddingLength = 8 - ((5 + payload.length) % 8);
+  if (paddingLength < 4) paddingLength += 8;
+  const padding = crypto.getRandomValues(new Uint8Array(paddingLength));
+  const packetLength = 1 + payload.length + paddingLength;
+  const out = new Uint8Array(4 + packetLength);
+  new DataView(out.buffer).setUint32(0, packetLength, false);
+  out[4] = paddingLength;
+  out.set(payload, 5);
+  out.set(padding, 5 + payload.length);
+  return out;
+}
+
+async function readPacket(conn: Deno.Conn, deadline: number): Promise<Uint8Array> {
+  const header = await readExactly(conn, 4, deadline);
+  const packetLength = new DataView(header.buffer, header.byteOffset, 4).getUint32(0, false);
+  if (packetLength < 1 || packetLength > 1 << 20) {
+    throw new KeyscanError(`implausible SSH packet length ${packetLength}`, "protocol_error");
+  }
+  const rest = await readExactly(conn, packetLength, deadline);
+  const paddingLength = rest[0];
+  const payload = rest.subarray(1, rest.length - paddingLength);
+  return payload;
+}
+
+function buildKexInitPayload(): Uint8Array {
+  const cookie = crypto.getRandomValues(new Uint8Array(16));
+  return new SshWriter()
+    .writeRaw(new Uint8Array([SSH_MSG_KEXINIT]))
+    .writeRaw(cookie)
+    .writeNameList(["curve25519-sha256"])
+    .writeNameList(["ssh-ed25519"])
+    .writeNameList(["aes128-ctr"])
+    .writeNameList(["aes128-ctr"])
+    .writeNameList(["hmac-sha2-256"])
+    .writeNameList(["hmac-sha2-256"])
+    .writeNameList(["none"])
+    .writeNameList(["none"])
+    .writeNameList([])
+    .writeNameList([])
+    .writeRaw(new Uint8Array([0])) // first_kex_packet_follows = false
+    .writeUint32(0) // reserved
+    .toBytes();
+}
+
+function parseServerKexInit(payload: Uint8Array): { serverHostKeyAlgorithms: string[] } {
+  const reader = new SshReader(payload.subarray(1)); // skip message type byte
+  reader.readBytes(16); // cookie
+  reader.readUtf8String(); // kex_algorithms
+  const serverHostKeyAlgorithms = reader.readUtf8String().split(",");
+  return { serverHostKeyAlgorithms };
+}
+
+export async function scanHostKey(
+  hostname: string,
+  port: number,
+  options: { timeoutMs?: number } = {},
+): Promise<ScannedHostKey> {
+  const timeoutMs = options.timeoutMs ?? 8000;
+  const deadline = Date.now() + timeoutMs;
+
+  let conn: Deno.Conn;
+  try {
+    conn = await Deno.connect({ hostname, port, transport: "tcp" });
+  } catch (err) {
+    throw new KeyscanError(`could not connect to ${hostname}:${port}: ${(err as Error).message}`, "connect_failed");
+  }
+
+  try {
+    // Identification string exchange (RFC 4253 section 4.2).
+    const serverId = await readLine(conn, deadline);
+    if (!serverId.startsWith("SSH-2.0-") && !serverId.startsWith("SSH-1.99-")) {
+      throw new KeyscanError(`unexpected SSH identification string: '${serverId}'`, "protocol_error");
+    }
+    await conn.write(new TextEncoder().encode("SSH-2.0-TreqControlPlaneKeyscan_1.0\r\n"));
+
+    // Key exchange init.
+    await conn.write(framePacket(buildKexInitPayload()));
+    const serverKexInit = await readPacket(conn, deadline);
+    if (serverKexInit[0] !== SSH_MSG_KEXINIT) {
+      throw new KeyscanError("expected SSH_MSG_KEXINIT from server", "protocol_error");
+    }
+    const { serverHostKeyAlgorithms } = parseServerKexInit(serverKexInit);
+    if (!serverHostKeyAlgorithms.includes("ssh-ed25519")) {
+      throw new KeyscanError(
+        `server does not offer ssh-ed25519 as a host key algorithm (offered: ${serverHostKeyAlgorithms.join(", ")})`,
+        "unsupported_host_key_algorithm",
+      );
+    }
+
+    // ECDH init with a throwaway 32-byte client value: we only need the
+    // reply's host key field, not a completed shared secret (see module
+    // doc-comment for why this is a deliberate, documented limitation).
+    const clientEphemeral = crypto.getRandomValues(new Uint8Array(32));
+    const ecdhInitPayload = new SshWriter()
+      .writeRaw(new Uint8Array([SSH_MSG_KEX_ECDH_INIT]))
+      .writeString(clientEphemeral)
+      .toBytes();
+    await conn.write(framePacket(ecdhInitPayload));
+
+    // The server may resend KEXINIT-adjacent packets; scan forward until the
+    // ECDH reply, bounded by the overall deadline.
+    let replyPayload: Uint8Array | null = null;
+    while (Date.now() < deadline) {
+      const packet = await readPacket(conn, deadline);
+      if (packet[0] === SSH_MSG_KEX_ECDH_REPLY) {
+        replyPayload = packet;
+        break;
+      }
+    }
+    if (!replyPayload) throw new KeyscanError("timed out waiting for SSH_MSG_KEX_ECDH_REPLY", "timeout");
+
+    const reader = new SshReader(replyPayload.subarray(1));
+    const hostKeyBlob = reader.readString();
+    // Q_S and signature follow but are not needed to record the host key.
+
+    const line = `ssh-ed25519 ${base64Encode(hostKeyBlob)}`;
+    const parsed: ParsedPublicKey = await parseOpenSshPublicKey(line);
+    return {
+      algorithm: parsed.algorithm,
+      fingerprintSha256: parsed.fingerprintSha256,
+      publicKeyLine: line,
+    };
+  } finally {
+    try {
+      conn.close();
+    } catch {
+      // already closed by the peer - nothing to do.
+    }
+  }
+}
diff --git a/supabase/functions/_shared/remote/ssh-vm-config.ts b/supabase/functions/_shared/remote/ssh-vm-config.ts
new file mode 100644
index 00000000..8bd9b7a3
--- /dev/null
+++ b/supabase/functions/_shared/remote/ssh-vm-config.ts
@@ -0,0 +1,76 @@
+// Shell command generators for configuring SSH trust on an already-booted
+// managed VM, mirroring `core::remote_bootstrap` in src-tauri. These are run
+// through `ManagedComputeProvider.execOnMachine`, not interpolated from
+// frontend text (PRD "Do not interpolate frontend text into remote shell
+// scripts") - every value here is server-generated: a CA public key line or
+// an already-validated OpenSSH public key line.
+
+const MANAGED_SSH_USER_HOME = "/home/treq";
+const TRUSTED_CA_MARKER_FILE = "/etc/ssh/treq_ca.pub";
+const SSHD_CONFIG_DROPIN = "/etc/ssh/sshd_config.d/60-treq-ca.conf";
+
+// Idempotent: writes the CA public key line to a fixed path and points
+// `TrustedUserCAKeys` at it via an sshd config drop-in, then reloads sshd.
+// Safe to re-run (e.g. on every reprovision) since it only overwrites its own
+// two files.
+export function installCaTrustCommand(caPublicKeyLine: string): string[] {
+  const script = `#!/bin/sh
+set -eu
+mkdir -p /etc/ssh/sshd_config.d
+cat > "${TRUSTED_CA_MARKER_FILE}" <<'TREQ_CA_EOF'
+${caPublicKeyLine}
+TREQ_CA_EOF
+cat > "${SSHD_CONFIG_DROPIN}" <<'TREQ_SSHD_EOF'
+TrustedUserCAKeys ${TRUSTED_CA_MARKER_FILE}
+TREQ_SSHD_EOF
+if command -v systemctl >/dev/null 2>&1 && systemctl is-active sshd >/dev/null 2>&1; then
+  systemctl reload sshd
+elif command -v systemctl >/dev/null 2>&1 && systemctl is-active ssh >/dev/null 2>&1; then
+  systemctl reload ssh
+elif [ -f /var/run/sshd.pid ]; then
+  kill -HUP "$(cat /var/run/sshd.pid)"
+fi
+echo "treq: CA trust installed"
+`;
+  return ["/bin/sh", "-c", script];
+}
+
+// Idempotent authorized_keys install: appends the key only if a line with
+// the same fingerprint marker comment is not already present, so repeated
+// installs (retry, re-registration) never duplicate an entry.
+export function installAuthorizedKeyCommand(publicKeyLine: string, fingerprintSha256: string): string[] {
+  const marker = `# treq-client-key:${fingerprintSha256}`;
+  const script = `#!/bin/sh
+set -eu
+mkdir -p "${MANAGED_SSH_USER_HOME}/.ssh"
+chmod 700 "${MANAGED_SSH_USER_HOME}/.ssh"
+touch "${MANAGED_SSH_USER_HOME}/.ssh/authorized_keys"
+if ! grep -qF "${marker}" "${MANAGED_SSH_USER_HOME}/.ssh/authorized_keys" 2>/dev/null; then
+  printf '%s\\n%s\\n' "${marker}" "${publicKeyLine}" >> "${MANAGED_SSH_USER_HOME}/.ssh/authorized_keys"
+fi
+chmod 600 "${MANAGED_SSH_USER_HOME}/.ssh/authorized_keys"
+echo "treq: authorized key installed"
+`;
+  return ["/bin/sh", "-c", script];
+}
+
+// Removes exactly the two lines (marker + key) this module's install added
+// for a given fingerprint, leaving every other entry untouched.
+export function removeAuthorizedKeyCommand(fingerprintSha256: string): string[] {
+  const marker = `# treq-client-key:${fingerprintSha256}`;
+  const script = `#!/bin/sh
+set -eu
+FILE="${MANAGED_SSH_USER_HOME}/.ssh/authorized_keys"
+if [ -f "$FILE" ]; then
+  MARKER="${marker}"
+  awk -v marker="$MARKER" '
+    $0 == marker { skip = 2; next }
+    skip > 0 { skip--; next }
+    { print }
+  ' "$FILE" > "$FILE.treq_tmp"
+  mv "$FILE.treq_tmp" "$FILE"
+fi
+echo "treq: authorized key removed"
+`;
+  return ["/bin/sh", "-c", script];
+}
diff --git a/supabase/functions/_shared/remote/ssh-wire.ts b/supabase/functions/_shared/remote/ssh-wire.ts
new file mode 100644
index 00000000..dbe698f4
--- /dev/null
+++ b/supabase/functions/_shared/remote/ssh-wire.ts
@@ -0,0 +1,123 @@
+// Minimal SSH wire-format primitives (RFC 4251 section 5), shared by
+// certificate signing (ssh-cert.ts) and host-key scanning (ssh-keyscan.ts).
+// No SSH library dependency: these are the handful of encode/decode rules
+// (string, uint32, uint64, mpint) actually needed for those two jobs.
+
+export class SshWriter {
+  private chunks: Uint8Array[] = [];
+
+  writeUint32(value: number): this {
+    const buf = new Uint8Array(4);
+    new DataView(buf.buffer).setUint32(0, value >>> 0, false);
+    this.chunks.push(buf);
+    return this;
+  }
+
+  writeUint64(value: bigint): this {
+    const buf = new Uint8Array(8);
+    new DataView(buf.buffer).setBigUint64(0, value, false);
+    this.chunks.push(buf);
+    return this;
+  }
+
+  writeRaw(bytes: Uint8Array): this {
+    this.chunks.push(bytes);
+    return this;
+  }
+
+  writeString(bytes: Uint8Array | string): this {
+    const data = typeof bytes === "string" ? new TextEncoder().encode(bytes) : bytes;
+    this.writeUint32(data.length);
+    this.chunks.push(data);
+    return this;
+  }
+
+  // Encodes a list of strings as one SSH "string" field: NUL-joined entries
+  // wrapped in an outer length-prefixed string, per PROTOCOL.certkeys'
+  // "valid principals" / extensions encoding.
+  writeNameList(names: string[]): this {
+    const joined = names.join("\0");
+    this.writeString(joined);
+    return this;
+  }
+
+  // Positive mpint per RFC 4251: a leading 0x00 byte is prepended when the
+  // high bit of the first byte would otherwise be set, so the value cannot be
+  // misread as negative.
+  writeMpintFromUnsigned(bytes: Uint8Array): this {
+    let value = bytes;
+    if (value.length > 0 && (value[0] & 0x80) !== 0) {
+      const padded = new Uint8Array(value.length + 1);
+      padded.set(value, 1);
+      value = padded;
+    }
+    this.writeString(value);
+    return this;
+  }
+
+  toBytes(): Uint8Array {
+    const total = this.chunks.reduce((sum, c) => sum + c.length, 0);
+    const out = new Uint8Array(total);
+    let offset = 0;
+    for (const chunk of this.chunks) {
+      out.set(chunk, offset);
+      offset += chunk.length;
+    }
+    return out;
+  }
+}
+
+export class SshReader {
+  private offset = 0;
+  constructor(private readonly data: Uint8Array) {}
+
+  get remaining(): number {
+    return this.data.length - this.offset;
+  }
+
+  readUint32(): number {
+    const view = new DataView(this.data.buffer, this.data.byteOffset + this.offset, 4);
+    this.offset += 4;
+    return view.getUint32(0, false);
+  }
+
+  readBytes(length: number): Uint8Array {
+    const out = this.data.subarray(this.offset, this.offset + length);
+    this.offset += length;
+    return out;
+  }
+
+  readString(): Uint8Array {
+    const length = this.readUint32();
+    return this.readBytes(length);
+  }
+
+  readUtf8String(): string {
+    return new TextDecoder().decode(this.readString());
+  }
+}
+
+export function base64Encode(bytes: Uint8Array): string {
+  let binary = "";
+  for (const byte of bytes) binary += String.fromCharCode(byte);
+  return btoa(binary);
+}
+
+export function base64Decode(value: string): Uint8Array {
+  const binary = atob(value);
+  const out = new Uint8Array(binary.length);
+  for (let i = 0; i < binary.length; i++) out[i] = binary.charCodeAt(i);
+  return out;
+}
+
+export async function sha256(bytes: Uint8Array): Promise<Uint8Array> {
+  const digest = await crypto.subtle.digest("SHA-256", bytes as BufferSource);
+  return new Uint8Array(digest);
+}
+
+// OpenSSH's `SHA256:<base64-no-padding>` fingerprint format.
+export async function sshFingerprintSha256(keyBlob: Uint8Array): Promise<string> {
+  const digest = await sha256(keyBlob);
+  const b64 = base64Encode(digest).replace(/=+$/, "");
+  return `SHA256:${b64}`;
+}
diff --git a/supabase/functions/_shared/remote/stub-sprites-adapter.ts b/supabase/functions/_shared/remote/stub-sprites-adapter.ts
index d661a065..12f514d8 100644
--- a/supabase/functions/_shared/remote/stub-sprites-adapter.ts
+++ b/supabase/functions/_shared/remote/stub-sprites-adapter.ts
@@ -5,6 +5,7 @@
 
 import type {
   CreateInstanceParams,
+  MachineExecResult,
   ManagedComputeProvider,
   ManagedInstanceState,
   ProviderInstance,
@@ -75,6 +76,14 @@ export class StubSpritesProvider implements ManagedComputeProvider {
     machines.delete(providerId);
     return Promise.resolve();
   }
+
+  execOnMachine(providerId: string, _command: string[], _timeoutSeconds?: number): Promise<MachineExecResult> {
+    if (!machines.has(providerId)) throw new ProviderError("not_found", "stub instance not found");
+    // Local/service-qa stub: there is no real machine to run a command on, so
+    // this records success without touching anything, matching the rest of
+    // this adapter's "warm in-memory fake" behavior.
+    return Promise.resolve({ exitCode: 0, stdout: "", stderr: "" });
+  }
 }
 
 function toProviderInstance(machine: StubMachine): ProviderInstance {
diff --git a/supabase/functions/remote-instance/index.ts b/supabase/functions/remote-instance/index.ts
index 8206f2de..915ec68e 100644
--- a/supabase/functions/remote-instance/index.ts
+++ b/supabase/functions/remote-instance/index.ts
@@ -37,10 +37,14 @@ import {
   findExistingOperation,
   getInstanceForOwner,
   previousHostKeyFingerprint,
+  recordEndpointHostKey,
   recordManagedEndpoint,
   updateInstance,
   type InstanceRow,
 } from "../_shared/remote/instance-store.ts";
+import { KeyscanError, scanHostKey } from "../_shared/remote/ssh-keyscan.ts";
+import { caKeyMaterialFromEnv, caPublicKeyLine } from "../_shared/remote/ssh-cert.ts";
+import { installCaTrustCommand } from "../_shared/remote/ssh-vm-config.ts";
 
 const corsHeaders = {
   "Access-Control-Allow-Origin": "*",
@@ -160,6 +164,99 @@ async function handleStatus(supabase: SupabaseClient, ownerUserId: string): Prom
   return json({ instance, endpoint });
 }
 
+// Closes the Phase 2 "host key fingerprint not yet available" gap with a
+// real scan, and installs CA trust on the freshly (re)provisioned VM (PRD
+// "Configure the managed VM to trust the Treq SSH CA"). Both steps are best
+// effort at this point in provisioning: a fresh machine's sshd may not be
+// reachable for a few seconds after the provider reports it started, so a
+// failure here is recorded as an auditable readiness-stage failure rather
+// than failing the whole provision/reprovision operation - the caller (or a
+// later explicit `keyscan_endpoint` / retry) can complete it once the VM is
+// actually reachable.
+async function establishSshTrust(
+  supabase: SupabaseClient,
+  provider: ManagedComputeProvider,
+  params: {
+    ownerUserId: string;
+    instanceId: string;
+    endpointId: string;
+    providerResourceId: string;
+    hostname: string;
+    port: number;
+    generation: number;
+  },
+): Promise<void> {
+  if (isSpritesStubEnabled()) {
+    // The stub adapter's address (`stub-xxx.stub.internal`) is not a real
+    // reachable host: there is nothing to scan or exec against in local/
+    // service-qa mode. Record a clearly-labeled stub fingerprint so
+    // downstream code paths that expect a host key row still have one.
+    await recordEndpointHostKey(supabase, {
+      ownerUserId: params.ownerUserId,
+      endpointId: params.endpointId,
+      algorithm: "ssh-ed25519",
+      fingerprintSha256: "SHA256:stub-mode-no-real-host-key",
+      generation: params.generation,
+    });
+    await recordAuditEvent(supabase, {
+      ownerUserId: params.ownerUserId,
+      instanceId: params.instanceId,
+      endpointId: params.endpointId,
+      eventType: "host_key_registered",
+      detail: { note: "REMOTE_SPRITES_STUB active: recorded a placeholder fingerprint, not a real scan", generation: params.generation },
+    });
+    return;
+  }
+
+  try {
+    const scanned = await scanHostKey(params.hostname, params.port);
+    await recordEndpointHostKey(supabase, {
+      ownerUserId: params.ownerUserId,
+      endpointId: params.endpointId,
+      algorithm: scanned.algorithm,
+      fingerprintSha256: scanned.fingerprintSha256,
+      generation: params.generation,
+    });
+    await recordAuditEvent(supabase, {
+      ownerUserId: params.ownerUserId,
+      instanceId: params.instanceId,
+      endpointId: params.endpointId,
+      eventType: "host_key_registered",
+      detail: { algorithm: scanned.algorithm, fingerprint: scanned.fingerprintSha256, generation: params.generation },
+    });
+  } catch (err) {
+    const kind = err instanceof KeyscanError ? err.kind : "other";
+    await recordAuditEvent(supabase, {
+      ownerUserId: params.ownerUserId,
+      instanceId: params.instanceId,
+      endpointId: params.endpointId,
+      eventType: "readiness_stage_failed",
+      detail: { stage: "host_keyscan", reason: (err as Error).message, kind },
+    });
+  }
+
+  try {
+    const ca = caKeyMaterialFromEnv();
+    const result = await provider.execOnMachine(params.providerResourceId, installCaTrustCommand(caPublicKeyLine(ca)));
+    if (result.exitCode !== 0) throw new Error(`ca trust install exited ${result.exitCode}: ${result.stderr || result.stdout}`);
+    await recordAuditEvent(supabase, {
+      ownerUserId: params.ownerUserId,
+      instanceId: params.instanceId,
+      endpointId: params.endpointId,
+      eventType: "ca_trust_installed",
+      detail: { generation: params.generation },
+    });
+  } catch (err) {
+    await recordAuditEvent(supabase, {
+      ownerUserId: params.ownerUserId,
+      instanceId: params.instanceId,
+      endpointId: params.endpointId,
+      eventType: "ca_trust_install_failed",
+      detail: { error: (err as Error).message },
+    });
+  }
+}
+
 function requireIdempotencyKey(body: Record<string, unknown>): string {
   const key = body.idempotency_key;
   if (typeof key !== "string" || key.length === 0) {
@@ -256,15 +353,14 @@ async function handleEnsure(
         existingEndpointId: null,
       });
       await updateInstance(supabase, instance.id, { endpoint_id: endpointId });
-      await recordAuditEvent(supabase, {
+      await establishSshTrust(supabase, provider, {
         ownerUserId,
         instanceId: instance.id,
         endpointId,
-        eventType: "host_key_registered",
-        detail: {
-          note: "host key fingerprint not yet available from provider create response; recorded once obtained through a trusted provisioning path",
-          generation: 0,
-        },
+        providerResourceId: providerInstance.providerResourceId,
+        hostname: providerInstance.address,
+        port: MANAGED_SSH_PORT,
+        generation: 0,
       });
     } else {
       await recordAuditEvent(supabase, {
@@ -435,11 +531,24 @@ async function handleReprovision(
       const previousFingerprint = instance.endpoint_id
         ? await previousHostKeyFingerprint(supabase, instance.endpoint_id)
         : null;
-      // Host key material is not yet returned by the vendor create/replace
-      // response (see remote-instance/index.ts handleEnsure comment); this
-      // records the rotation slot so Phase 3 verification has old/new
-      // fingerprint + generation to compare once a real fingerprint is
-      // available.
+
+      // Real keyscan against the (possibly replaced) VM, recorded at the new
+      // generation, plus CA trust re-install (a replacement VM starts from
+      // the base image and does not inherit the previous machine's sshd
+      // config). This is the explicit host-key rotation record the PRD's
+      // "Reprovisioning may rotate the host key" paragraph calls for: old
+      // fingerprint, new fingerprint, generation, timestamp, and provider
+      // resource id are all captured here (initiating principal is
+      // `ownerUserId`, the only principal that can call reprovision).
+      await establishSshTrust(supabase, provider, {
+        ownerUserId,
+        instanceId: instance.id,
+        endpointId,
+        providerResourceId: providerInstance.providerResourceId,
+        hostname: providerInstance.address,
+        port: MANAGED_SSH_PORT,
+        generation: nextGeneration,
+      });
       await recordAuditEvent(supabase, {
         ownerUserId,
         instanceId: instance.id,
@@ -449,6 +558,7 @@ async function handleReprovision(
           previous_fingerprint: previousFingerprint,
           generation: nextGeneration,
           provider_resource_id: providerInstance.providerResourceId,
+          initiating_principal: ownerUserId,
         },
       });
     }
diff --git a/supabase/functions/remote-ssh-trust/index.ts b/supabase/functions/remote-ssh-trust/index.ts
new file mode 100644
index 00000000..35b84e44
--- /dev/null
+++ b/supabase/functions/remote-ssh-trust/index.ts
@@ -0,0 +1,511 @@
+// Edge function: SSH trust and authentication for Remote SSH Control
+// (prds/remote-ssh.md, Phase 3: "SSH trust and authentication").
+//
+// POST body: { action, idempotency_key?, ...action-specific fields }
+// action:
+//   "register_client_key"    - register a user-selected SSH public key (public material only)
+//   "list_client_keys"       - list the caller's registered keys
+//   "revoke_client_key"      - independently revoke one key
+//   "issue_certificate"      - sign a short-lived OpenSSH user certificate for a managed instance
+//   "install_authorized_key" - direct-key alternative: install a key into the managed VM's authorized_keys
+//   "remove_authorized_key"  - remove a previously installed authorized key
+//   "keyscan_endpoint"       - re-run the host-key scan against the caller's managed endpoint
+//
+// Auth: user JWT in Authorization header. Every action re-derives ownership
+// from `owner_user_id` server-side; a caller-supplied instance_id/key_id is
+// only ever used to look up a row already scoped to that owner (PRD
+// "Security requirements": never trust client-supplied IDs alone).
+//
+// The SSH CA private key lives only in this function's environment
+// (REMOTE_SSH_CA_ED25519_SEED_BASE64) - it is read to sign a certificate and
+// never included in any response or database write.
+
+import { createClient, type SupabaseClient } from "@supabase/supabase-js";
+import { recordAuditEvent } from "../_shared/remote/audit.ts";
+import {
+  beginOperation,
+  completeOperation,
+  findExistingOperation,
+  getInstanceById,
+  type InstanceRow,
+} from "../_shared/remote/instance-store.ts";
+import {
+  findClientKeyByFingerprint,
+  getOwnedClientKey,
+  insertClientKey,
+  listClientKeys,
+  recordAuthorizedKeyInstalled,
+  recordAuthorizedKeyRemoved,
+  revokeClientKey,
+  type ClientKeyRow,
+} from "../_shared/remote/client-key-store.ts";
+import { extractEd25519RawKey, parseOpenSshPublicKey, UnsupportedKeyError } from "../_shared/remote/ssh-keys.ts";
+import { caKeyMaterialFromEnv, issueEd25519UserCertificate, randomSerial } from "../_shared/remote/ssh-cert.ts";
+import { installAuthorizedKeyCommand, removeAuthorizedKeyCommand } from "../_shared/remote/ssh-vm-config.ts";
+import {
+  ProviderError,
+  SpritesProvider,
+  spritesConfigFromEnv,
+  type ManagedComputeProvider,
+} from "../_shared/remote/sprites-adapter.ts";
+import { isSpritesStubEnabled, StubSpritesProvider } from "../_shared/remote/stub-sprites-adapter.ts";
+import { KeyscanError, scanHostKey } from "../_shared/remote/ssh-keyscan.ts";
+import { recordEndpointHostKey } from "../_shared/remote/instance-store.ts";
+
+const corsHeaders = {
+  "Access-Control-Allow-Origin": "*",
+  "Access-Control-Allow-Headers": "authorization, x-client-info, apikey, content-type",
+};
+
+// Certificates are deliberately short-lived (PRD "Certificate lifetime
+// should be short enough to bound loss exposure while allowing normal
+// reconnects") while still tolerating a slow client clock or a long-running
+// interactive session started just before expiry.
+const CERTIFICATE_VALIDITY_MINUTES = 20;
+const CERTIFICATE_CLOCK_SKEW_MINUTES = 2;
+
+function json(body: unknown, status = 200): Response {
+  return new Response(JSON.stringify(body), {
+    status,
+    headers: { ...corsHeaders, "Content-Type": "application/json" },
+  });
+}
+
+class ValidationError extends Error {
+  constructor(message: string, public readonly status = 400) {
+    super(message);
+  }
+}
+
+function getProvider(): ManagedComputeProvider {
+  if (isSpritesStubEnabled()) return new StubSpritesProvider();
+  return new SpritesProvider(spritesConfigFromEnv());
+}
+
+function requireIdempotencyKey(body: Record<string, unknown>): string {
+  const key = body.idempotency_key;
+  if (typeof key !== "string" || key.length === 0) throw new ValidationError("idempotency_key is required");
+  return key;
+}
+
+function requireString(body: Record<string, unknown>, field: string): string {
+  const value = body[field];
+  if (typeof value !== "string" || value.length === 0) throw new ValidationError(`${field} is required`);
+  return value;
+}
+
+Deno.serve(async (req) => {
+  if (req.method === "OPTIONS") return new Response(null, { headers: corsHeaders, status: 204 });
+  if (req.method !== "POST") return json({ error: "Method not allowed" }, 405);
+
+  const authHeader = req.headers.get("authorization") ?? "";
+  const userToken = authHeader.replace(/^Bearer\s+/i, "");
+  if (!userToken) return json({ error: "Unauthorized" }, 401);
+
+  const supabaseUrl = Deno.env.get("SUPABASE_URL") ?? "";
+  const supabaseAnonKey = Deno.env.get("SUPABASE_ANON_KEY") ?? "";
+  const supabaseServiceKey = Deno.env.get("SUPABASE_SERVICE_ROLE_KEY") ?? "";
+
+  const supabaseUser = createClient(supabaseUrl, supabaseAnonKey, {
+    global: { headers: { Authorization: `Bearer ${userToken}` } },
+  });
+  const {
+    data: { user },
+    error: authError,
+  } = await supabaseUser.auth.getUser();
+  if (authError || !user) return json({ error: "Unauthorized" }, 401);
+
+  // deno-lint-ignore no-explicit-any
+  let body: Record<string, any>;
+  try {
+    body = await req.json();
+  } catch {
+    return json({ error: "Invalid JSON" }, 400);
+  }
+
+  const action = body.action;
+  const supabase = createClient(supabaseUrl, supabaseServiceKey);
+
+  try {
+    switch (action) {
+      case "register_client_key":
+        return await handleRegisterClientKey(supabase, user.id, body);
+      case "list_client_keys":
+        return await handleListClientKeys(supabase, user.id);
+      case "revoke_client_key":
+        return await handleRevokeClientKey(supabase, user.id, body);
+      case "issue_certificate":
+        return await handleIssueCertificate(supabase, user.id, body);
+      case "install_authorized_key":
+        return await handleAuthorizedKeyChange(supabase, user.id, body, "install");
+      case "remove_authorized_key":
+        return await handleAuthorizedKeyChange(supabase, user.id, body, "remove");
+      case "keyscan_endpoint":
+        return await handleKeyscanEndpoint(supabase, user.id, body);
+      default:
+        return json({ error: `Unknown action '${action}'` }, 400);
+    }
+  } catch (err) {
+    if (err instanceof ValidationError) return json({ error: err.message }, err.status);
+    if (err instanceof UnsupportedKeyError) return json({ error: err.message }, 400);
+    if (err instanceof ProviderError) return json({ error: err.message, provider_error: err.kind }, 502);
+    if (err instanceof KeyscanError) return json({ error: err.message, keyscan_error: err.kind }, 502);
+    console.error(`remote-ssh-trust action=${action} failed: ${(err as Error).message}`);
+    return json({ error: "Internal error" }, 500);
+  }
+});
+
+function toClientKeyResponse(row: ClientKeyRow) {
+  return {
+    id: row.id,
+    algorithm: row.algorithm,
+    fingerprint_sha256: row.fingerprint_sha256,
+    comment: row.comment,
+    created_at: row.created_at,
+    revoked_at: row.revoked_at,
+  };
+}
+
+// Register a user-selected public key. Only public material, its algorithm,
+// and its fingerprint are stored (PRD "Client key policy") - the request
+// body never carries a private key and none is generated here.
+async function handleRegisterClientKey(
+  supabase: SupabaseClient,
+  ownerUserId: string,
+  // deno-lint-ignore no-explicit-any
+  body: Record<string, any>,
+): Promise<Response> {
+  const idempotencyKey = requireIdempotencyKey(body);
+  const publicKeyLine = requireString(body, "public_key");
+  const comment: string | null = typeof body.comment === "string" ? body.comment : null;
+
+  const existingOp = await findExistingOperation(supabase, ownerUserId, idempotencyKey);
+  if (existingOp) {
+    const keys = await listClientKeys(supabase, ownerUserId);
+    return json({ operation_id: existingOp.id, status: existingOp.status, keys: keys.map(toClientKeyResponse) });
+  }
+
+  const parsed = await parseOpenSshPublicKey(publicKeyLine);
+
+  const op = await beginOperation(supabase, {
+    ownerUserId,
+    instanceId: null,
+    operationType: "register_client_key",
+    idempotencyKey,
+  });
+
+  const existing = await findClientKeyByFingerprint(supabase, ownerUserId, parsed.fingerprintSha256);
+  if (existing && !existing.revoked_at) {
+    await completeOperation(supabase, op.id, { status: "succeeded" });
+    return json({ operation_id: op.id, status: "succeeded", key: toClientKeyResponse(existing) });
+  }
+
+  const row = await insertClientKey(supabase, {
+    ownerUserId,
+    publicKey: publicKeyLine,
+    algorithm: parsed.algorithm,
+    fingerprintSha256: parsed.fingerprintSha256,
+    comment: comment ?? parsed.comment,
+  });
+  await completeOperation(supabase, op.id, { status: "succeeded" });
+  await recordAuditEvent(supabase, {
+    ownerUserId,
+    eventType: "client_key_registered",
+    detail: { key_id: row.id, algorithm: row.algorithm, fingerprint: row.fingerprint_sha256 },
+  });
+  return json({ operation_id: op.id, status: "succeeded", key: toClientKeyResponse(row) });
+}
+
+async function handleListClientKeys(supabase: SupabaseClient, ownerUserId: string): Promise<Response> {
+  const keys = await listClientKeys(supabase, ownerUserId);
+  return json({ keys: keys.map(toClientKeyResponse) });
+}
+
+// Each client key is independently revocable (PRD Goal 6/7). Revoking does
+// not remove any authorized_keys installs already made with it - callers
+// that also want the VM entry removed should call remove_authorized_key.
+async function handleRevokeClientKey(
+  supabase: SupabaseClient,
+  ownerUserId: string,
+  // deno-lint-ignore no-explicit-any
+  body: Record<string, any>,
+): Promise<Response> {
+  const idempotencyKey = requireIdempotencyKey(body);
+  const keyId = requireString(body, "key_id");
+
+  const existingOp = await findExistingOperation(supabase, ownerUserId, idempotencyKey);
+  if (existingOp) return json({ operation_id: existingOp.id, status: existingOp.status });
+
+  const key = await getOwnedClientKey(supabase, ownerUserId, keyId);
+  if (!key) throw new ValidationError("Key does not belong to this user", 404);
+
+  const op = await beginOperation(supabase, {
+    ownerUserId,
+    instanceId: null,
+    operationType: "revoke_client_key",
+    idempotencyKey,
+  });
+  await revokeClientKey(supabase, ownerUserId, keyId);
+  await completeOperation(supabase, op.id, { status: "succeeded" });
+  await recordAuditEvent(supabase, {
+    ownerUserId,
+    eventType: "client_key_revoked",
+    detail: { key_id: keyId, fingerprint: key.fingerprint_sha256 },
+  });
+  return json({ operation_id: op.id, status: "succeeded" });
+}
+
+// Resolves and verifies an owned, ready managed instance with its endpoint,
+// per the "verifies ownership, key status, and instance status" step of the
+// PRD's certificate flow.
+async function requireReadyOwnedInstance(
+  supabase: SupabaseClient,
+  ownerUserId: string,
+  instanceId: string,
+): Promise<InstanceRow> {
+  const instance = await getInstanceById(supabase, ownerUserId, instanceId);
+  if (!instance) throw new ValidationError("Instance does not belong to this user", 404);
+  if (instance.status !== "ready") {
+    throw new ValidationError(`Instance is not ready (status: ${instance.status})`, 409);
+  }
+  if (!instance.endpoint_id) throw new ValidationError("Instance has no endpoint recorded yet", 409);
+  return instance;
+}
+
+async function requireActiveOwnedKey(
+  supabase: SupabaseClient,
+  ownerUserId: string,
+  keyId: string,
+): Promise<ClientKeyRow> {
+  const key = await getOwnedClientKey(supabase, ownerUserId, keyId);
+  if (!key) throw new ValidationError("Key does not belong to this user", 404);
+  if (key.revoked_at) throw new ValidationError("Key has been revoked", 409);
+  return key;
+}
+
+// Signs a short-lived OpenSSH user certificate for a managed instance (PRD
+// "Managed VM certificate flow", steps 4-5). The CA private key is read from
+// an Edge Function secret for the duration of this call only.
+async function handleIssueCertificate(
+  supabase: SupabaseClient,
+  ownerUserId: string,
+  // deno-lint-ignore no-explicit-any
+  body: Record<string, any>,
+): Promise<Response> {
+  const instanceId = requireString(body, "instance_id");
+  const keyId = requireString(body, "key_id");
+
+  const instance = await requireReadyOwnedInstance(supabase, ownerUserId, instanceId);
+  const key = await requireActiveOwnedKey(supabase, ownerUserId, keyId);
+
+  if (key.algorithm !== "ssh-ed25519") {
+    // Real gap, not a stub: only ed25519 user keys are certified today (see
+    // ssh-cert.ts doc comment). RSA/ECDSA certificate signing is future work.
+    throw new ValidationError(
+      `Certificate signing is only implemented for ssh-ed25519 keys (this key is ${key.algorithm})`,
+      400,
+    );
+  }
+
+  const { data: endpointRow, error: endpointError } = await supabase
+    .from("remote_endpoints")
+    .select("id, hostname, port, username, source, instance_id")
+    .eq("id", instance.endpoint_id)
+    .eq("owner_user_id", ownerUserId)
+    .maybeSingle();
+  if (endpointError) throw new Error(`failed to read endpoint: ${endpointError.message}`);
+  if (!endpointRow) throw new ValidationError("Endpoint does not belong to this user", 404);
+
+  const { data: hostKeyRows, error: hostKeyError } = await supabase
+    .from("remote_endpoint_host_keys")
+    .select("algorithm, fingerprint_sha256, comment")
+    .eq("endpoint_id", endpointRow.id)
+    .is("revoked_at", null)
+    .order("generation", { ascending: false });
+  if (hostKeyError) throw new Error(`failed to read host keys: ${hostKeyError.message}`);
+
+  const parsedUserKey = await parseOpenSshPublicKey(key.public_key);
+  const rawUserKey = extractEd25519RawKey(parsedUserKey);
+
+  const ca = caKeyMaterialFromEnv();
+  const now = new Date();
+  const validAfter = new Date(now.getTime() - CERTIFICATE_CLOCK_SKEW_MINUTES * 60_000);
+  const validBefore = new Date(now.getTime() + CERTIFICATE_VALIDITY_MINUTES * 60_000);
+  const serial = randomSerial();
+
+  let issued;
+  try {
+    issued = await issueEd25519UserCertificate({
+      ca,
+      userPublicKey: rawUserKey,
+      principals: [endpointRow.username, instance.id],
+      serial,
+      keyId: `treq:${ownerUserId}:${instance.id}`,
+      validAfter,
+      validBefore,
+    });
+  } catch (err) {
+    await recordAuditEvent(supabase, {
+      ownerUserId,
+      instanceId: instance.id,
+      endpointId: endpointRow.id,
+      eventType: "certificate_issue_failed",
+      detail: { error: (err as Error).message },
+    });
+    throw err;
+  }
+
+  await recordAuditEvent(supabase, {
+    ownerUserId,
+    instanceId: instance.id,
+    endpointId: endpointRow.id,
+    eventType: "certificate_issued",
+    detail: {
+      serial: issued.serial,
+      principals: [endpointRow.username, instance.id],
+      issued_at: validAfter.toISOString(),
+      expires_at: validBefore.toISOString(),
+      key_id: key.id,
+    },
+  });
+
+  return json({
+    certificate: issued.certificateLine,
+    serial: issued.serial,
+    expires_at: validBefore.toISOString(),
+    endpoint: {
+      id: endpointRow.id,
+      instance_id: endpointRow.instance_id,
+      source: { type: "managed", provider: "fly_sprites", generation: instance.generation },
+      hostname: endpointRow.hostname,
+      port: endpointRow.port,
+      username: endpointRow.username,
+      host_keys: (hostKeyRows ?? []).map((row: { algorithm: string; fingerprint_sha256: string; comment: string | null }) => ({
+        algorithm: row.algorithm,
+        fingerprint_sha256: row.fingerprint_sha256,
+        comment: row.comment,
+      })),
+      authentication: { type: "certificate", key_reference: key.id },
+    },
+  });
+}
+
+// Direct existing-key auth alternative (PRD "Existing keys without
+// certificates"): idempotently installs or removes a registered public key
+// in the managed VM's authorized_keys via the provider's exec channel.
+async function handleAuthorizedKeyChange(
+  supabase: SupabaseClient,
+  ownerUserId: string,
+  // deno-lint-ignore no-explicit-any
+  body: Record<string, any>,
+  mode: "install" | "remove",
+): Promise<Response> {
+  const idempotencyKey = requireIdempotencyKey(body);
+  const instanceId = requireString(body, "instance_id");
+  const keyId = requireString(body, "key_id");
+
+  const existingOp = await findExistingOperation(supabase, ownerUserId, idempotencyKey);
+  if (existingOp) return json({ operation_id: existingOp.id, status: existingOp.status });
+
+  const instance = await requireReadyOwnedInstance(supabase, ownerUserId, instanceId);
+  const key = await getOwnedClientKey(supabase, ownerUserId, keyId);
+  if (!key) throw new ValidationError("Key does not belong to this user", 404);
+  if (mode === "install" && key.revoked_at) throw new ValidationError("Key has been revoked", 409);
+  if (!instance.provider_resource_id) throw new ValidationError("Instance has no provider resource yet", 409);
+
+  const op = await beginOperation(supabase, {
+    ownerUserId,
+    instanceId: instance.id,
+    operationType: mode === "install" ? "install_authorized_key" : "remove_authorized_key",
+    idempotencyKey,
+  });
+
+  try {
+    const provider = getProvider();
+    const command =
+      mode === "install"
+        ? installAuthorizedKeyCommand(key.public_key, key.fingerprint_sha256)
+        : removeAuthorizedKeyCommand(key.fingerprint_sha256);
+    const result = await provider.execOnMachine(instance.provider_resource_id, command);
+    if (result.exitCode !== 0) {
+      throw new Error(`authorized_keys ${mode} exited ${result.exitCode}: ${result.stderr || result.stdout}`);
+    }
+
+    if (mode === "install") {
+      await recordAuthorizedKeyInstalled(supabase, {
+        ownerUserId,
+        endpointId: instance.endpoint_id!,
+        clientKeyId: key.id,
+      });
+    } else {
+      await recordAuthorizedKeyRemoved(supabase, instance.endpoint_id!, key.id);
+    }
+
+    await completeOperation(supabase, op.id, { status: "succeeded" });
+    await recordAuditEvent(supabase, {
+      ownerUserId,
+      instanceId: instance.id,
+      endpointId: instance.endpoint_id,
+      eventType: mode === "install" ? "authorized_key_installed" : "authorized_key_removed",
+      detail: { key_id: key.id, fingerprint: key.fingerprint_sha256 },
+    });
+    return json({ operation_id: op.id, status: "succeeded" });
+  } catch (err) {
+    await completeOperation(supabase, op.id, { status: "failed", errorMessage: (err as Error).message });
+    throw err;
+  }
+}
+
+// Re-runs the real host-key scan against the caller's managed endpoint and
+// records old/new fingerprint, generation, and provider resource id as a
+// rotation record (PRD "Reprovisioning may rotate the host key"). Also used
+// by remote-instance's ensure/reprovision flow immediately after an address
+// becomes available, closing the Phase 2 host-key gap.
+async function handleKeyscanEndpoint(
+  supabase: SupabaseClient,
+  ownerUserId: string,
+  // deno-lint-ignore no-explicit-any
+  body: Record<string, any>,
+): Promise<Response> {
+  const instanceId = requireString(body, "instance_id");
+  const instance = await getInstanceById(supabase, ownerUserId, instanceId);
+  if (!instance) throw new ValidationError("Instance does not belong to this user", 404);
+  if (!instance.endpoint_id) throw new ValidationError("Instance has no endpoint recorded yet", 409);
+
+  const { data: endpointRow, error: endpointError } = await supabase
+    .from("remote_endpoints")
+    .select("id, hostname, port")
+    .eq("id", instance.endpoint_id)
+    .eq("owner_user_id", ownerUserId)
+    .maybeSingle();
+  if (endpointError) throw new Error(`failed to read endpoint: ${endpointError.message}`);
+  if (!endpointRow) throw new ValidationError("Endpoint does not belong to this user", 404);
+
+  try {
+    const scanned = await scanHostKey(endpointRow.hostname, endpointRow.port);
+    await recordEndpointHostKey(supabase, {
+      ownerUserId,
+      endpointId: endpointRow.id,
+      algorithm: scanned.algorithm,
+      fingerprintSha256: scanned.fingerprintSha256,
+      generation: instance.generation,
+    });
+    await recordAuditEvent(supabase, {
+      ownerUserId,
+      instanceId: instance.id,
+      endpointId: endpointRow.id,
+      eventType: "host_key_registered",
+      detail: { algorithm: scanned.algorithm, fingerprint: scanned.fingerprintSha256, generation: instance.generation },
+    });
+    return json({ host_key: { algorithm: scanned.algorithm, fingerprint_sha256: scanned.fingerprintSha256 } });
+  } catch (err) {
+    await recordAuditEvent(supabase, {
+      ownerUserId,
+      instanceId: instance.id,
+      endpointId: endpointRow.id,
+      eventType: "host_keyscan_failed",
+      detail: { error: (err as Error).message },
+    });
+    throw err;
+  }
+}
diff --git a/supabase/migrations/016_remote_ssh_trust.sql b/supabase/migrations/016_remote_ssh_trust.sql
new file mode 100644
index 00000000..df6b2ed6
--- /dev/null
+++ b/supabase/migrations/016_remote_ssh_trust.sql
@@ -0,0 +1,54 @@
+-- Remote SSH trust and authentication (Phase 3: prds/remote-ssh.md
+-- "SSH trust and authentication"). Adds the columns and tables needed for
+-- client key algorithm tracking, direct authorized_keys installs, and the
+-- wider set of operation types this phase introduces. The SSH CA private key
+-- itself is never stored here - only server-side Edge Function secrets ever
+-- hold it (see supabase/functions/_shared/remote/ssh-cert.ts).
+
+-- remote_client_keys previously stored only the raw public key text and its
+-- fingerprint; certificate signing needs to know the algorithm up front
+-- (only ssh-ed25519 is certifiable today) without re-parsing the key on
+-- every read.
+alter table public.remote_client_keys
+  add column algorithm text not null default 'ssh-ed25519';
+
+alter table public.remote_client_keys
+  alter column algorithm drop default;
+
+-- Direct-key authentication alternative (PRD "Existing keys without
+-- certificates"): tracks which of a user's registered client keys have been
+-- installed into which endpoint's authorized_keys, so install/remove is
+-- idempotent and auditable rather than a fire-and-forget shell append.
+create table public.remote_endpoint_authorized_keys (
+  id uuid primary key default gen_random_uuid(),
+  owner_user_id uuid not null references auth.users(id) on delete cascade,
+  endpoint_id uuid not null references public.remote_endpoints(id) on delete cascade,
+  client_key_id uuid not null references public.remote_client_keys(id) on delete cascade,
+  installed_at timestamptz not null default now(),
+  removed_at timestamptz,
+  unique (endpoint_id, client_key_id)
+);
+
+alter table public.remote_endpoint_authorized_keys enable row level security;
+
+create policy "Users can manage own authorized key installs"
+  on public.remote_endpoint_authorized_keys for all
+  using (owner_user_id = auth.uid())
+  with check (owner_user_id = auth.uid());
+
+create index remote_endpoint_authorized_keys_endpoint_idx
+  on public.remote_endpoint_authorized_keys (endpoint_id);
+
+-- Phase 3 adds authorized-key install/remove and an explicit host-keyscan
+-- operation type alongside the Phase 1 set.
+alter table public.remote_instance_operations
+  drop constraint remote_instance_operations_operation_type_check;
+
+alter table public.remote_instance_operations
+  add constraint remote_instance_operations_operation_type_check
+  check (operation_type in (
+    'provision', 'wake', 'reprovision', 'delete',
+    'register_client_key', 'revoke_client_key', 'issue_certificate',
+    'register_endpoint', 'register_repository',
+    'install_authorized_key', 'remove_authorized_key', 'keyscan_host_key'
+  ));