security: contain physical miner commands in Stage A

Disable physical control through legacy HTTP/M30 producers, typed admission,
queue dispatch and upgraded Windows executors. Restrict diagnostic transports
to exact read-only operations, with ordinal PowerShell validation before sockets.

Retire unsupported pending/claimed work with atomic, idempotent security audits.
Preserve prior delivery evidence and report unknown outcomes after a prior claim,
including explicitly repaired pending rows. Include containment, cancellation,
concurrency and malformed-payload regressions and the old-Agent cutover runbook.

Validation: independent final application-security revalidation PASS. Two unchanged
baseline source-assertion failures are documented as exceptions. Production release
still requires native Windows validation and verified fencing/stop/drain/upgrade;
server cancellation alone cannot revoke a cached command held by an old Agent.

agents/windows/EnrichOneMiner.ps1

@@ -23,6 +23,24 @@ function Invoke-MinerCommandRaw {
         [int]$TimeoutMs = 2500
     )
 
+    # This transport accepts only the exact queries used by this enrichment script.
+    # Validate before creating a socket; callers cannot reinterpret a write as a read.
+    $readCommands = @("summary", "devs", "edevs", "stats", "pools", "devdetails",
+        "version", "get_version", "get_miner_info")
+    # PowerShell containment operators use culture comparisons, even with -c.
+    # Match ordinally and serialize the canonical literal; never trim/normalize.
+    $approvedCommand = $null
+    foreach ($candidate in $readCommands) {
+        if ([string]::Equals($candidate, $Command, [StringComparison]::Ordinal)) {
+            $approvedCommand = $candidate
+            break
+        }
+    }
+    if ($Port -ne 4028 -or $null -eq $approvedCommand) {
+        throw "unsupported_diagnostic_operation_stage_a"
+    }
+    $requestJson = @{ cmd = $approvedCommand } | ConvertTo-Json -Compress
+
     $client = $null
 
     try {
@@ -38,7 +56,7 @@ function Invoke-MinerCommandRaw {
         $client.SendTimeout = $TimeoutMs
 
         $stream = $client.GetStream()
-        $payload = [System.Text.Encoding]::ASCII.GetBytes("{`"cmd`":`"$Command`"}`n")
+        $payload = [System.Text.Encoding]::ASCII.GetBytes($requestJson + "`n")
         $stream.Write($payload, 0, $payload.Length)
 
         try {

agents/windows/MinerOpsAgentService.ps1

@@ -743,61 +743,7 @@ function Invoke-MinerPlainCommand {
 
         [int]$TimeoutMs = 2500
     )
-
-    $client = $null
-
-    try {
-        $client = New-Object System.Net.Sockets.TcpClient
-        $async = $client.BeginConnect($IpAddress, $Port, $null, $null)
-
-        if (-not $async.AsyncWaitHandle.WaitOne($TimeoutMs)) {
-            throw "Connect timeout to $IpAddress`:$Port"
-        }
-
-        $client.EndConnect($async)
-        $client.ReceiveTimeout = $TimeoutMs
-        $client.SendTimeout = $TimeoutMs
-
-        $stream = $client.GetStream()
-        $payload = [System.Text.Encoding]::ASCII.GetBytes("{`"cmd`":`"$Command`"}`n")
-        $stream.Write($payload, 0, $payload.Length)
-
-        try {
-            $client.Client.Shutdown([System.Net.Sockets.SocketShutdown]::Send)
-        } catch {
-        }
-
-        $buffer = New-Object byte[] 65536
-        $chunks = New-Object System.Collections.ArrayList
-        $started = Get-Date
-
-        while (((Get-Date) - $started).TotalMilliseconds -lt $TimeoutMs) {
-            if (-not $stream.DataAvailable) {
-                Start-Sleep -Milliseconds 20
-                continue
-            }
-
-            $read = $stream.Read($buffer, 0, $buffer.Length)
-
-            if ($read -le 0) {
-                break
-            }
-
-            for ($i = 0; $i -lt $read; $i++) {
-                [void]$chunks.Add([byte]$buffer[$i])
-            }
-
-            if ($chunks.Count -gt 1048576) {
-                break
-            }
-        }
-
-        return [System.Text.Encoding]::UTF8.GetString([byte[]]$chunks.ToArray([byte])).Trim()
-    } finally {
-        if ($client -ne $null) {
-            $client.Close()
-        }
-    }
+    throw "physical_control_disabled_stage_a"
 }
 
 function Get-MinerOpsApi3Credential {
@@ -894,61 +840,7 @@ function Invoke-MinerOpsApi3Payload {
 
         [int]$TimeoutMs = 10000
     )
-
-    $client = $null
-    $started = Get-Date
-
-    try {
-        $client = New-Object System.Net.Sockets.TcpClient
-
-        $async = $client.BeginConnect($IpAddress, $Port, $null, $null)
-
-        if (-not $async.AsyncWaitHandle.WaitOne($TimeoutMs)) {
-            throw "API3 connect timeout to $IpAddress`:$Port"
-        }
-
-        $client.EndConnect($async)
-        $client.ReceiveTimeout = $TimeoutMs
-        $client.SendTimeout = $TimeoutMs
-
-        $json = $Payload | ConvertTo-Json -Depth 20 -Compress
-        $requestBytes = [System.Text.Encoding]::UTF8.GetBytes($json)
-        $lengthBytes = [System.BitConverter]::GetBytes([UInt32]$requestBytes.Length)
-
-        $stream = $client.GetStream()
-        $stream.Write($lengthBytes, 0, $lengthBytes.Length)
-        $stream.Write($requestBytes, 0, $requestBytes.Length)
-
-        $responseLengthBytes = Read-MinerOpsExactBytes -Stream $stream -Length 4 -TimeoutMs $TimeoutMs
-        $responseLength = [System.BitConverter]::ToUInt32($responseLengthBytes, 0)
-
-        if ($responseLength -le 0 -or $responseLength -gt 1048576) {
-            throw "API3 response length looked invalid: $responseLength"
-        }
-
-        $responseBytes = Read-MinerOpsExactBytes -Stream $stream -Length ([int]$responseLength) -TimeoutMs $TimeoutMs
-        $responseText = [System.Text.Encoding]::UTF8.GetString($responseBytes)
-        $responseJson = $responseText | ConvertFrom-Json
-
-        return @{
-            sent = $true
-            protocol = "whatsminer_api3"
-            target_ip = $IpAddress
-            target_port = $Port
-            duration_seconds = [math]::Round(((Get-Date) - $started).TotalSeconds, 3)
-            request = @{
-                cmd = $Payload.cmd
-                param = $Payload.param
-                account = $Payload.account
-                has_token = [bool]$Payload.token
-            }
-            response = $responseJson
-        }
-    } finally {
-        if ($client -ne $null) {
-            $client.Close()
-        }
-    }
+    throw "physical_control_disabled_stage_a"
 }
 
 function Get-MinerOpsApi3Salt {
@@ -989,43 +881,7 @@ function Invoke-MinerOpsApi3Service {
         [ValidateSet("start", "stop", "restart")]
         [string]$ServiceAction
     )
-
-    $account = Get-MinerOpsApi3Credential -Name "api3_account" -DefaultValue ""
-    $password = Get-MinerOpsApi3Credential -Name "api3_password" -DefaultValue ""
-    $portRaw = Get-MinerOpsApi3Credential -Name "api3_port" -DefaultValue "4433"
-
-    try {
-        $port = [int]$portRaw
-    } catch {
-        $port = 4433
-    }
-
-    $command = "set.miner.service"
-    $saltInfo = Get-MinerOpsApi3Salt -IpAddress $IpAddress -Port $port
-    $salt = [string]$saltInfo.salt
-    $timestamp = [int64][DateTimeOffset]::UtcNow.ToUnixTimeSeconds()
-    $token = Get-MinerOpsApi3Token -Command $command -Password $password -Salt $salt -Timestamp $timestamp
-
-    $payload = @{
-        cmd = $command
-        param = $ServiceAction
-        ts = $timestamp
-        token = $token
-        account = $account
-    }
-
-    $result = Invoke-MinerOpsApi3Payload -IpAddress $IpAddress -Port $port -Payload $payload
-
-    $result["salt_response"] = $saltInfo.salt_response
-    $result["api3_command"] = $command
-    $result["api3_param"] = $ServiceAction
-    $result["api3_account"] = $account
-
-    if ([int]$result.response.code -ne 0) {
-        throw "API3 $command $ServiceAction failed: $($result.response | ConvertTo-Json -Compress)"
-    }
-
-    return $result
+    throw "physical_control_disabled_stage_a"
 }
 
 
@@ -1054,45 +910,7 @@ function Invoke-MinerOpsApi3AuthenticatedCommand {
 
         [object]$Param = $null
     )
-
-    $account = Get-MinerOpsApi3Credential -Name "api3_account" -DefaultValue ""
-    $password = Get-MinerOpsApi3Credential -Name "api3_password" -DefaultValue ""
-    $portRaw = Get-MinerOpsApi3Credential -Name "api3_port" -DefaultValue "4433"
-
-    try {
-        $port = [int]$portRaw
-    } catch {
-        $port = 4433
-    }
-
-    $saltInfo = Get-MinerOpsApi3Salt -IpAddress $IpAddress -Port $port
-    $salt = [string]$saltInfo.salt
-    $timestamp = [int64][DateTimeOffset]::UtcNow.ToUnixTimeSeconds()
-    $token = Get-MinerOpsApi3Token -Command $Command -Password $password -Salt $salt -Timestamp $timestamp
-
-    $payload = @{
-        cmd = $Command
-        ts = $timestamp
-        token = $token
-        account = $account
-    }
-
-    if ($null -ne $Param) {
-        $payload["param"] = $Param
-    }
-
-    $result = Invoke-MinerOpsApi3Payload -IpAddress $IpAddress -Port $port -Payload $payload
-
-    $result["salt_response"] = $saltInfo.salt_response
-    $result["api3_command"] = $Command
-    $result["api3_param"] = $Param
-    $result["api3_account"] = $account
-
-    if ([int]$result.response.code -ne 0) {
-        throw "API3 $Command failed: $($result.response | ConvertTo-Json -Compress)"
-    }
-
-    return $result
+    throw "physical_control_disabled_stage_a"
 }
 
 function Get-MinerOpsPayloadValue {
@@ -1129,113 +947,14 @@ function Invoke-MinerOpsLegacy4028Command {
         [int] $Port = 4028,
         [int] $TimeoutMs = 7000
     )
-
-    $normalizedIp = Normalize-MinerOpsIpAddress -IpAddress $IpAddress
-
-    $client = New-Object System.Net.Sockets.TcpClient
-    $connectResult = $client.BeginConnect($normalizedIp, $Port, $null, $null)
-
-    if (-not $connectResult.AsyncWaitHandle.WaitOne($TimeoutMs, $false)) {
-        try { $client.Close() } catch {}
-        throw "Legacy 4028 connection timed out to $normalizedIp`:$Port"
-    }
-
-    $client.EndConnect($connectResult)
-
-    try {
-        $stream = $client.GetStream()
-        $stream.ReadTimeout = $TimeoutMs
-        $stream.WriteTimeout = $TimeoutMs
-
-        $bytes = [System.Text.Encoding]::ASCII.GetBytes($CommandJson)
-        $stream.Write($bytes, 0, $bytes.Length)
-        $stream.Flush()
-
-        Start-Sleep -Milliseconds 500
-
-        $buffer = New-Object byte[] 65535
-        $builder = New-Object System.Text.StringBuilder
-
-        do {
-            $read = $stream.Read($buffer, 0, $buffer.Length)
-
-            if ($read -gt 0) {
-                [void]$builder.Append([System.Text.Encoding]::ASCII.GetString($buffer, 0, $read))
-            }
-        } while ($stream.DataAvailable)
-
-        $rawResponse = $builder.ToString().Trim([char]0).Trim()
-
-        $parsedResponse = $null
-
-        try {
-            if ($rawResponse) {
-                $parsedResponse = $rawResponse | ConvertFrom-Json
-            }
-        } catch {
-            $parsedResponse = $null
-        }
-
-        return @{
-            success = $true
-            control_method = "legacy_4028"
-            port = $Port
-            request_json = $CommandJson
-            raw_response = $rawResponse
-            parsed_response = $parsedResponse
-        }
-    }
-    finally {
-        try { $stream.Close() } catch {}
-        try { $client.Close() } catch {}
-    }
+    throw "physical_control_disabled_stage_a"
 }
 
 function Invoke-MinerOpsM30Legacy4028Test {
     param(
         [Parameter(Mandatory=$true)] $Payload
     )
-
-    if ($Payload -is [string]) {
-        $Payload = $Payload | ConvertFrom-Json
-    }
-
-    $ipAddress = Get-MinerOpsPayloadValue -Payload $Payload -Name "ip_address"
-
-    if (-not $ipAddress) {
-        $ipAddress = Get-MinerOpsPayloadValue -Payload $Payload -Name "current_ip"
-    }
-
-    if (-not $ipAddress) {
-        throw "Legacy 4028 test failed: no IP address in payload."
-    }
-
-    $legacyJson = Get-MinerOpsPayloadValue -Payload $Payload -Name "legacy_4028_json"
-    $legacyCommand = Get-MinerOpsPayloadValue -Payload $Payload -Name "legacy_4028_command"
-
-    if (-not $legacyJson) {
-        if (-not $legacyCommand) {
-            throw "Legacy 4028 test failed: no legacy command in payload."
-        }
-
-        $legacyJson = '{"command":"' + $legacyCommand + '"}'
-    }
-
-    Write-AgentLog "Running legacy 4028 M30 test against $ipAddress command=$legacyCommand"
-
-    $result = Invoke-MinerOpsLegacy4028Command -IpAddress $ipAddress -CommandJson $legacyJson -Port 4028
-
-    return @{
-        success = $true
-        action = "legacy_4028"
-        test_lab = "m30"
-        ip_address = $ipAddress
-        legacy_4028_command = $legacyCommand
-        legacy_4028_request = $legacyJson
-        legacy_4028_response = $result.raw_response
-        legacy_4028_parsed = $result.parsed_response
-        control_method = "legacy_4028"
-    }
+    throw "physical_control_disabled_stage_a"
 }
 
 function Run-MinerControlCommand {
@@ -1244,105 +963,7 @@ function Run-MinerControlCommand {
         [Parameter(Mandatory=$true)]
         [object]$Payload
     )
-    # MinerOps M30 legacy 4028 test path
-    if ($Payload -is [string]) {
-        try {
-            $Payload = $Payload | ConvertFrom-Json
-        } catch {
-        }
-    }
-
-    if ($Payload -and (($Payload.control_method -eq "legacy_4028") -or ($Payload.action -eq "legacy_4028"))) {
-        return Invoke-MinerOpsM30Legacy4028Test -Payload $Payload
-    }
-
-
-    if ($Payload -is [string]) {
-        $Payload = $Payload | ConvertFrom-Json
-    }
-
-    if ($null -eq $Payload) {
-        throw "Miner control payload was empty."
-    }
-
-    $action = ([string]$Payload.action).ToLower()
-    $minerId = [string]$Payload.miner_id
-    $macAddress = [string]$Payload.mac_address
-    $dryRun = [bool]$Payload.dry_run
-
-    $ipAddress = ([string]$Payload.ip_address).Trim()
-
-    if ($ipAddress.Contains("/")) {
-        $ipAddress = $ipAddress.Split("/")[0]
-    }
-
-    if ([string]::IsNullOrWhiteSpace($ipAddress)) {
-        throw "Miner control payload did not include ip_address."
-    }
-
-    $serviceAction = ""
-    $api3Command = ""
-    $api3Param = $null
-    $performanceMode = ""
-
-    if ($action -eq "reboot") {
-        $serviceAction = "restart"
-        $api3Command = "set.miner.service"
-    } elseif ($action -eq "suspend") {
-        $serviceAction = "stop"
-        $api3Command = "set.miner.service"
-    } elseif ($action -eq "resume") {
-        $serviceAction = "start"
-        $api3Command = "set.miner.service"
-    } elseif ($action -eq "performance_mode") {
-        $performanceMode = ([string]$Payload.performance_mode).ToLower()
-
-        if ($performanceMode -in @("low", "normal", "high")) {
-            $api3Command = "set.miner.power_mode"
-            $api3Param = $performanceMode
-        } else {
-            throw "Invalid performance_mode: $performanceMode"
-        }
-    } else {
-        throw "Unsupported miner control action: $action"
-    }
-
-    Write-AgentLog "Miner control requested. Action=$action API3Command=$api3Command API3Service=$serviceAction PerformanceMode=$performanceMode IP=$ipAddress MAC=$macAddress DryRun=$dryRun"
-
-    if ($dryRun) {
-        return @{
-            ok = $true
-            dry_run = $true
-            action = $action
-            api3_command = $api3Command
-            api3_service_action = $serviceAction
-            performance_mode = $performanceMode
-            ip_address = $ipAddress
-            miner_id = $minerId
-            mac_address = $macAddress
-            message = "Dry run only. No miner command sent."
-        }
-    }
-
-    if ($action -eq "performance_mode") {
-        $api3Result = Invoke-MinerOpsApi3AuthenticatedCommand -IpAddress $ipAddress -Command $api3Command -Param $api3Param
-    } else {
-        $api3Result = Invoke-MinerOpsApi3Service -IpAddress $ipAddress -ServiceAction $serviceAction
-    }
-
-    return @{
-        ok = $true
-        dry_run = $false
-        action = $action
-        api3_command = $api3Command
-        api3_service_action = $serviceAction
-        performance_mode = $performanceMode
-        ip_address = $ipAddress
-        miner_id = $minerId
-        mac_address = $macAddress
-        message = "API3 miner control command sent by Windows Agent."
-        api3_result = $api3Result
-    }
+    throw "physical_control_disabled_stage_a"
 }
 
 
@@ -1368,8 +989,8 @@ function Process-PendingCommands {
                 Write-AgentLog "Command completed: $commandId"
             } elseif ($commandType -eq "miner_control") {
                 $result = Run-MinerControlCommand -Payload $cmd.request_payload
-                Send-AgentCommandResult -CommandId $commandId -Status "completed" -ResultPayload $result | Out-Null
-                Write-AgentLog "Miner control command completed: $commandId"
+                Send-AgentCommandResult -CommandId $commandId -Status "failed" -ResultPayload $result | Out-Null
+                Write-AgentLog "Miner control rejected by Stage A containment: $commandId"
 
             } elseif ($commandType -eq "enrich_match_discovery") {
                 $payloadObj = $cmd.request_payload

agents/windows/MinerOpsApi3EnrichmentPoller.ps1

@@ -30,11 +30,57 @@ function Invoke-WhatsMinerApi3Read {
         [int] $TimeoutMs = 2500
     )
 
+    # Exact operation and parameter allowlist at the shared transport boundary.
+    # No extra keys (including credentials, command aliases, or parameter overrides)
+    # reach the wire, even when this function is called outside the poller loop.
+    $readCommands = @("get.device.info", "get.miner.setting", "get.miner.status",
+        "get.fan.setting", "get.device.custom_data")
+    $command = $Request["cmd"]
+    $approvedCommand = $null
+    if ($command -is [string]) {
+        foreach ($candidate in $readCommands) {
+            if ([string]::Equals($candidate, $command, [StringComparison]::Ordinal)) {
+                $approvedCommand = $candidate
+                break
+            }
+        }
+    }
+    if ($null -eq $approvedCommand) {
+        throw "unsupported_diagnostic_operation_stage_a"
+    }
+    $expectedKeys = @("cmd")
+    $wireRequest = [ordered]@{ cmd = $approvedCommand }
+    if ([string]::Equals($approvedCommand, "get.miner.status", [StringComparison]::Ordinal)) {
+        $expectedKeys += "param"
+        if ($Request["param"] -isnot [string] -or
+            -not [string]::Equals($Request["param"], "summary+pools+edevs", [StringComparison]::Ordinal)) {
+            throw "unsupported_diagnostic_operation_stage_a"
+        }
+        $wireRequest["param"] = "summary+pools+edevs"
+    }
+    if ($Request.Count -ne $expectedKeys.Count) {
+        throw "unsupported_diagnostic_operation_stage_a"
+    }
+    foreach ($key in $Request.Keys) {
+        $approvedKey = $false
+        if ($key -is [string]) {
+            foreach ($expectedKey in $expectedKeys) {
+                if ([string]::Equals($expectedKey, $key, [StringComparison]::Ordinal)) {
+                    $approvedKey = $true
+                    break
+                }
+            }
+        }
+        if (-not $approvedKey) {
+            throw "unsupported_diagnostic_operation_stage_a"
+        }
+    }
+
     $client = $null
     $stream = $null
 
     try {
-        $json = ($Request | ConvertTo-Json -Depth 20 -Compress)
+        $json = ($wireRequest | ConvertTo-Json -Depth 20 -Compress)
         $payload = [System.Text.Encoding]::ASCII.GetBytes($json)
         $length = [BitConverter]::GetBytes([Int32]$payload.Length)
 

agents/windows/MinerOpsControl.ps1

@@ -26,29 +26,8 @@ function Get-MinerOpsControlCredentials {
 }
 
 function Invoke-MinerOpsControlApi3Command {
-    param(
-        [Parameter(Mandatory=$true)][string]$IpAddress,
-        [Parameter(Mandatory=$true)][string]$Command,
-        [object]$Param = $null
-    )
-    $credentials = Get-MinerOpsControlCredentials
-    if (-not $credentials.available) { throw "credentials_missing" }
-
-    $saltInfo = Get-MinerOpsApi3Salt -IpAddress $IpAddress -Port $credentials.port
-    $timestamp = [int64][DateTimeOffset]::UtcNow.ToUnixTimeSeconds()
-    $token = Get-MinerOpsApi3Token -Command $Command -Password $credentials.password `
-        -Salt ([string]$saltInfo.salt) -Timestamp $timestamp
-    $payload = @{
-        cmd = $Command
-        ts = $timestamp
-        token = $token
-        account = $credentials.account
-    }
-    if ($null -ne $Param) { $payload["param"] = $Param }
-    $response = Invoke-MinerOpsApi3Payload -IpAddress $IpAddress `
-        -Port $credentials.port -Payload $payload
-    if ([int]$response.response.code -ne 0) { throw "command_rejected" }
-    return $response
+    param([string]$IpAddress, [string]$Command, [object]$Param = $null)
+    throw "physical_control_disabled_stage_a"
 }
 
 function New-MinerOpsControlResult {
@@ -74,90 +53,11 @@ function New-MinerOpsControlResult {
 
 function Run-MinerControlCommand {
     param([Parameter(Mandatory=$true)][object]$Payload)
-
-    try {
-        if ($Payload -is [string]) { $Payload = $Payload | ConvertFrom-Json }
-        $action = ([string]$Payload.action).Trim().ToLower()
-        $ipAddress = ([string]$Payload.ip_address).Trim()
-        $requested = ([string]$Payload.requested_value).Trim().ToLower()
-        if (-not $requested -and $Payload.performance_mode) {
-            $requested = ([string]$Payload.performance_mode).Trim().ToLower()
-        }
-        if ($action -notin @("suspend", "resume", "performance_mode")) {
-            return New-MinerOpsControlResult -Accepted $false `
-                -VerificationAttempted $false -VerificationStatus "unavailable" `
-                -FailureCode "invalid_action" -QueueStatus "failed"
-        }
-        if ([string]::IsNullOrWhiteSpace($ipAddress)) {
-            return New-MinerOpsControlResult -Accepted $false `
-                -VerificationAttempted $false -VerificationStatus "unavailable" `
-                -FailureCode "target_missing" -QueueStatus "failed"
-        }
-        $credentials = Get-MinerOpsControlCredentials
-        if (-not $credentials.available) {
-            return New-MinerOpsControlResult -Accepted $false `
-                -VerificationAttempted $false -VerificationStatus "unavailable" `
-                -FailureCode "credentials_missing" -QueueStatus "failed"
-        }
-
-        if ($action -eq "performance_mode") {
-            if ($requested -notin @("low", "normal", "high")) {
-                return New-MinerOpsControlResult -Accepted $false `
-                    -VerificationAttempted $false -VerificationStatus "unavailable" `
-                    -FailureCode "invalid_mode" -QueueStatus "failed"
-            }
-            Invoke-MinerOpsControlApi3Command -IpAddress $ipAddress `
-                -Command "set.miner.power_mode" -Param $requested | Out-Null
-
-            $observed = ""
-            $readUnavailable = $false
-            for ($attempt = 1; $attempt -le 3; $attempt++) {
-                try {
-                    $read = Invoke-MinerOpsControlApi3Command -IpAddress $ipAddress `
-                        -Command "get.miner.setting"
-                    $candidate = ([string]$read.response.msg.'power-mode').Trim().ToLower()
-                    if ($candidate -in @("low", "normal", "high")) {
-                        $observed = $candidate
-                        if ($observed -eq $requested) {
-                            return New-MinerOpsControlResult -Accepted $true `
-                                -VerificationAttempted $true -VerificationStatus "confirmed" `
-                                -ObservedMode $observed
-                        }
-                    }
-                } catch {
-                    if ($_.Exception.Message -match "unsupported|unknown_command") {
-                        $readUnavailable = $true
-                        break
-                    }
-                }
-                if ($attempt -lt 3) { Start-Sleep -Milliseconds 750 }
-            }
-            if ($readUnavailable) {
-                return New-MinerOpsControlResult -Accepted $true `
-                    -VerificationAttempted $true -VerificationStatus "unavailable"
-            }
-            if ($observed) {
-                return New-MinerOpsControlResult -Accepted $true `
-                    -VerificationAttempted $true `
-                    -VerificationStatus "verification_failed" -ObservedMode $observed
-            }
-            return New-MinerOpsControlResult -Accepted $true `
-                -VerificationAttempted $true -VerificationStatus "unconfirmed"
-        }
-
-        $serviceAction = if ($action -eq "suspend") { "stop" } else { "start" }
-        Invoke-MinerOpsControlApi3Command -IpAddress $ipAddress `
-            -Command "set.miner.service" -Param $serviceAction | Out-Null
-        return New-MinerOpsControlResult -Accepted $true `
-            -VerificationAttempted $false -VerificationStatus "unconfirmed"
-    } catch {
-        $failure = if ($_.Exception.Message -eq "credentials_missing") {
-            "credentials_missing"
-        } else { "write_rejected" }
-        return New-MinerOpsControlResult -Accepted $false `
-            -VerificationAttempted $false -VerificationStatus "unavailable" `
-            -FailureCode $failure -QueueStatus "failed"
-    }
+    # No current request schema binds authorization to physical identity.
+    # Do not parse, normalize, reinterpret, or transmit an old queued payload.
+    return New-MinerOpsControlResult -Accepted $false `
+        -VerificationAttempted $false -VerificationStatus "unavailable" `
+        -FailureCode "physical_control_disabled_stage_a" -QueueStatus "failed"
 }
 
 function Send-AgentCommandResult {
@@ -185,7 +85,6 @@ function Send-AgentCommandResult {
 
 function Send-AgentCheckin {
     $config = Read-AgentConfig
-    $credentials = Get-MinerOpsControlCredentials
     $body = [ordered]@{
         agent_id = [string]$config.agent_id
         site_id = [string]$config.site_id
@@ -193,8 +92,8 @@ function Send-AgentCheckin {
         hostname = $env:COMPUTERNAME
         version = Get-MinerOpsAgentVersion
         ip_address = ""
-        control_available = [bool]$credentials.available
-        control_unavailable_reason = if ($credentials.available) { $null } else { "credentials_missing" }
+        control_available = $false
+        control_unavailable_reason = "physical_control_disabled_stage_a"
         local_time = (Get-Date).ToString("o")
     }
     return Invoke-AgentApi -Method "POST" -Path "/api/agent/checkin" -Body $body

agents/windows/README.md

@@ -1,7 +1,44 @@
 # MinerOps Windows Agent
 
+## Stage A physical-control containment
+
+Physical miner control is disabled in this version, including suspend, resume,
+performance-mode changes, reboot, M30 tests, and miner-side password changes.
+No environment flag, configured API3 credential, or Agent capability report can
+enable it. This is containment, not completion of Blocker #3 identity binding.
+
+The Hub returns HTTP 410 for retired legacy command, scheduling, and unrestricted
+diagnostic endpoints (after authorization where applicable). Typed control POSTs
+require `miners.control` and return HTTP 409 `physical_control_disabled_stage_a`.
+The unrestricted `whatsminer-read-test` endpoint is not a supported read-only tool.
+
+Authenticated queue polling cancels pending/claimed physical and unknown command
+types, records an atomic `agent.command.rejected` audit entry, and only delivers
+fixed telemetry/discovery command types. Results cannot promote retired entries.
+Original payloads, prior results/errors, timestamps, and terminal history remain.
+Cancellation of an already claimed command does **not** prove it was never sent;
+the audit explicitly records its previous execution outcome as unknown.
+
+The Windows executor rejects all current physical-control schemas without any
+device connection and advertises physical control as unavailable. Existing Agents
+must be stopped/drained and upgraded before claiming fleet-wide containment:
+server cancellation cannot recall payloads cached by older running Agents.
+Do not enable control or deploy this work before separate security review.
+The ordered, fail-closed procedure is in
+[the Stage A old-Agent cutover runbook](../../docs/operations/STAGE_A_OLD_AGENT_CUTOVER.txt).
+Native Windows PowerShell 5.1/SYSTEM validation and verified fleet network fencing
+remain production gates; server cancellation alone cannot contain an old executor.
+
+Retained legacy/API3 enrichment helpers validate exact read commands and fixed
+parameters before opening a socket. Unknown operations and extra keys fail closed.
+Schedule cancel/delete require `miners.control` and retire matching live unsupported
+commands for the stored Agent with atomic actor-attributed rejection audits.
+Deletion retains schedule history; retries do not duplicate audits. Previously
+claimed execution remains unknown. Manual cutover retirement does not claim read
+work and must sweep offline/queue-only Agent IDs until the global live count is zero.
+
 The Windows Agent runs at a mining site and performs local discovery, polling,
-API3 enrichment, pool polling, and approved queued miner-control operations.
+API3 enrichment, and pool polling. Physical miner control is disabled in Stage A.
 MinerOps does not send API3 credentials through the hub command queue; those
 credentials remain in the Agent's local `agent-config.json`.
 

agents/windows/tests/Test-MinerOpsControl.ps1

@@ -1,84 +1,115 @@
 $ErrorActionPreference = "Stop"
-$script:config = [pscustomobject]@{ api3_account = "operator"; api3_password = "secret"; api3_port = 4433 }
-$script:calls = @()
-$script:outcomes = @()
-$script:outcomeIndex = 0
-
-function Read-AgentConfig { return $script:config }
-. (Join-Path (Split-Path $PSScriptRoot -Parent) "MinerOpsControl.ps1")
-
-function Invoke-MinerOpsControlApi3Command {
-    param([string]$IpAddress, [string]$Command, [object]$Param = $null)
-    $script:calls += [pscustomobject]@{ command = $Command; param = $Param }
-    $outcome = $script:outcomes[$script:outcomeIndex]
-    $script:outcomeIndex += 1
-    if ($outcome -is [System.Exception]) { throw $outcome }
-    return $outcome
-}
+$script:assertions = 0
+$script:physicalCalls = 0
+$script:readCalls = 0
+$script:apiCalls = @()
+$script:commands = @()
 
 function Assert-Equal($Actual, $Expected, [string]$Message) {
+    $script:assertions++
     if ($Actual -ne $Expected) { throw "$Message. Expected '$Expected', got '$Actual'." }
 }
-function Reset-Harness([object[]]$Outcomes) {
-    $script:calls = @()
-    $script:outcomes = @($Outcomes)
-    $script:outcomeIndex = 0
-}
-function Response([string]$Mode = "") {
-    return @{ response = @{ code = 0; msg = @{ "power-mode" = $Mode } } }
-}
-
-$payload = @{ action = "performance_mode"; requested_value = "normal"; ip_address = "192.0.2.1" }
-
-$script:config = [pscustomobject]@{ api3_account = ""; api3_password = ""; api3_port = 4433 }
-Reset-Harness @()
-$result = Run-MinerControlCommand -Payload $payload
-Assert-Equal $result.failure_code "credentials_missing" "Missing credentials code"
-Assert-Equal $script:calls.Count 0 "Missing credentials must not contact API3"
-
-$script:config = [pscustomobject]@{ api3_account = "operator"; api3_password = "secret"; api3_port = 4433 }
-Reset-Harness @((Response), (Response "normal"))
-$result = Run-MinerControlCommand -Payload $payload
-Assert-Equal $result.verification_status "confirmed" "Performance confirmation"
-Assert-Equal @($script:calls | Where-Object command -Like "set.*").Count 1 "Exactly one write"
-
-Reset-Harness @((Response), (Response "low"), (Response "low"), (Response "low"))
-$result = Run-MinerControlCommand -Payload $payload
-Assert-Equal $result.verification_status "verification_failed" "Contradictory mode"
-Assert-Equal $script:calls.Count 4 "One write and three bounded reads"
-
-Reset-Harness @((Response), (Response), (Response), (Response))
-$result = Run-MinerControlCommand -Payload $payload
-Assert-Equal $result.verification_status "unconfirmed" "Ambiguous reads"
-
-Reset-Harness @((Response), [System.Exception]::new("unknown_command"))
-$result = Run-MinerControlCommand -Payload $payload
-Assert-Equal $result.verification_status "unavailable" "Unavailable read path"
-
-foreach ($action in @("suspend", "resume")) {
-    Reset-Harness @((Response))
-    $result = Run-MinerControlCommand -Payload @{
-        action = $action; requested_value = $null; ip_address = "192.0.2.1"
+function Assert-Throws($Block, [string]$Message) {
+    $caught = $false
+    try { & $Block | Out-Null } catch {
+        $caught = $true
+        Assert-Equal $_.Exception.Message "physical_control_disabled_stage_a" $Message
     }
-    Assert-Equal $result.command_accepted $true "$action accepted"
-    Assert-Equal $result.verification_status "unconfirmed" "$action remains unconfirmed"
-    Assert-Equal $result.observed_service_state $null "$action has no invented state"
-    Assert-Equal @($script:calls | Where-Object command -Like "set.*").Count 1 "$action one write"
+    Assert-Equal $caught $true "$Message must reject"
 }
 
-Reset-Harness @([System.Exception]::new("vendor response included secret"))
-$result = Run-MinerControlCommand -Payload @{
-    action = "suspend"; requested_value = $null; ip_address = "192.0.2.1"
+# Define functions from the real service AST, without executing its service loop,
+# config loading, startup, or any scheduled tasks.
+$root = Split-Path $PSScriptRoot -Parent
+$tokens = $null; $parseErrors = $null
+$ast = [System.Management.Automation.Language.Parser]::ParseFile(
+    (Join-Path $root "MinerOpsAgentService.ps1"), [ref]$tokens, [ref]$parseErrors)
+Assert-Equal $parseErrors.Count 0 "Service parser"
+foreach ($function in $ast.FindAll({
+    param($node)
+    $node -is [System.Management.Automation.Language.FunctionDefinitionAst] -and
+        $node.Parent -is [System.Management.Automation.Language.NamedBlockAst]
+}, $true)) {
+    Invoke-Expression $function.Extent.Text
 }
-Assert-Equal $result.failure_code "write_rejected" "Write rejection is sanitized"
-$json = $result | ConvertTo-Json -Compress
-if ($json -match "vendor|secret|192\\.0\\.2\\.1") { throw "Typed result leaked raw data" }
 
-Reset-Harness @()
-$result = Run-MinerControlCommand -Payload @{
-    action = "performance_mode"; requested_value = "turbo"; ip_address = "192.0.2.1"
+$script:productionApi3Payload = (Get-Command Invoke-MinerOpsApi3Payload).ScriptBlock
+$script:productionPlainCommand = (Get-Command Invoke-MinerPlainCommand).ScriptBlock
+function Invoke-MinerOpsApi3Payload { $script:physicalCalls++; throw "Unexpected API3 transport" }
+function Get-MinerOpsApi3Salt { $script:physicalCalls++; throw "Unexpected salt probe" }
+function Invoke-MinerPlainCommand { $script:physicalCalls++; throw "Unexpected legacy transport" }
+function Read-AgentConfig { return [pscustomobject]@{
+    agent_id = "simulated-agent"; site_id = ""; site_name = "Simulation"
+    api3_account = "operator"; api3_password = "mock-only"; api3_port = 4433
+} }
+function Write-AgentLog { param([string]$Message) }
+function Invoke-AgentApi {
+    param([string]$Method, [string]$Path, [object]$Body)
+    $script:apiCalls += @{ method = $Method; path = $Path; body = $Body }
+    return @{ ok = $true }
 }
-Assert-Equal $result.failure_code "invalid_mode" "Invalid mode code"
-Assert-Equal $script:calls.Count 0 "Invalid mode must not contact API3"
+function Get-PendingAgentCommands { return $script:commands }
+function Run-KnownMinerPoll { $script:readCalls++; return @{ simulated = $true } }
+function Run-NewMinerDiscovery { $script:readCalls++; return @{ simulated = $true } }
 
-Write-Host "MinerOps typed Agent control tests passed."
+# The pre-module legacy implementation must also be inert, including helpers
+# which used to support direct generic API3 or M30/4028 writes.
+Assert-Throws { Run-MinerControlCommand -Payload @{ action = "reboot"; dry_run = $false } } "Legacy executor"
+Assert-Throws { Invoke-MinerOpsApi3Service -IpAddress "192.0.2.1" -ServiceAction "stop" } "Legacy service helper"
+Assert-Throws { & $script:productionApi3Payload -IpAddress "192.0.2.1" -Payload @{ cmd = "set.miner.service"; param = "stop" } } "Generic API3 payload helper"
+Assert-Throws { & $script:productionPlainCommand -IpAddress "192.0.2.1" -Command "reboot" } "Generic legacy command helper"
+Assert-Throws { Invoke-MinerOpsApi3AuthenticatedCommand -IpAddress "192.0.2.1" -Command "set.miner.service" -Param "stop" } "Legacy API3"
+Assert-Throws { Invoke-MinerOpsLegacy4028Command -IpAddress "192.0.2.1" -CommandJson '{"command":"reboot"}' } "Legacy 4028"
+Assert-Throws { Invoke-MinerOpsM30Legacy4028Test -Payload @{ legacy_4028_command = "reboot" } } "M30 helper"
+
+. (Join-Path $root "MinerOpsControl.ps1")
+Assert-Throws { Invoke-MinerOpsControlApi3Command -IpAddress "192.0.2.1" -Command "set.miner.service" -Param "stop" } "Typed transport"
+Assert-Throws { Invoke-MinerOpsControlApi3Command -IpAddress "192.0.2.1" -Command "get.read_then_write" } "Disguised diagnostic"
+
+$payloads = @(
+    @{ action = "legacy_4028"; legacy_4028_command = "reboot"; dry_run = $false },
+    @{ action = "reboot"; ip_address = "192.0.2.1" },
+    @{ schema_version = 1; action = "suspend"; ip_address = "192.0.2.1"; miner_id = "spoofed" },
+    @{ schema_version = 1; action = "resume"; ip_address = "192.0.2.1" },
+    @{ schema_version = 1; action = "performance_mode"; requested_value = "high"; ip_address = "192.0.2.1" },
+    @{ schema_version = 999; action = "suspend" },
+    @{ schema_version = "1"; action = "performance_mode"; performance_mode = "high" },
+    @{ command = "summary"; params = @{ cmd = "set.miner.service"; param = "stop" } },
+    @{ action = "read_reboot"; dry_run = $true },
+    @{},
+    '{"schema_version":0,"action":"reboot"}',
+    "{invalid-json"
+)
+foreach ($payload in $payloads) {
+    $result = Run-MinerControlCommand -Payload $payload
+    Assert-Equal $result.command_accepted $false "Rejected payload acceptance"
+    Assert-Equal $result.failure_code "physical_control_disabled_stage_a" "Rejected payload reason"
+    Assert-Equal $result._queue_status "failed" "Rejected payload terminal result"
+    Assert-Equal $result.verification_attempted $false "No physical verification/probe"
+    Assert-Equal $result.observed_service_state $null "No invented device state"
+    Assert-Equal $script:physicalCalls 0 "No physical calls"
+    $script:commands = @(@{ command_id = "legacy-delivery"; command_type = "miner_control"; request_payload = $payload })
+    $script:apiCalls = @()
+    Assert-Equal (Process-PendingCommands) 1 "Delivered legacy command handled"
+    Assert-Equal $script:apiCalls.Count 1 "One failed result submitted"
+    Assert-Equal $script:apiCalls[0].body.status "failed" "Legacy delivery failed"
+    Assert-Equal $script:apiCalls[0].body.result_payload.command_accepted $false "Legacy delivery not accepted"
+    Assert-Equal $script:physicalCalls 0 "No writes after delivery"
+}
+foreach ($kind in @("raw_api3", "miner_control_v2", "read_reboot", "")) {
+    $script:commands = @(@{ command_id = "unsupported"; command_type = $kind; request_payload = @{ action = "reboot" } })
+    $script:apiCalls = @()
+    Assert-Equal (Process-PendingCommands) 1 "Unknown command handled"
+    Assert-Equal $script:apiCalls[0].body.status "failed" "Unknown command fails closed"
+}
+Assert-Equal $script:readCalls 0 "No physical request reinterpreted as telemetry"
+$script:commands = @(@{ command_id = "read-only"; command_type = "poll_known_miners_now"; request_payload = @{} })
+$script:apiCalls = @()
+Assert-Equal (Process-PendingCommands) 1 "Fixed read-only command available"
+Assert-Equal $script:readCalls 1 "Only simulated read performed"
+Assert-Equal $script:physicalCalls 0 "No physical calls in entire suite"
+$script:apiCalls = @()
+Send-AgentCheckin | Out-Null
+Assert-Equal $script:apiCalls[0].body.control_available $false "Cannot advertise physical capability"
+Assert-Equal $script:apiCalls[0].body.control_unavailable_reason "physical_control_disabled_stage_a" "Containment advertised"
+Write-Host "Stage A Windows containment passed: $($script:assertions) assertions; zero physical transport calls."

agents/windows/tests/Test-MinerOpsDiagnosticTransport.ps1

@@ -0,0 +1,199 @@
+$ErrorActionPreference = "Stop"
+$script:assertions = 0
+$script:connections = 0
+$TimeoutMs = 1000
+
+function Assert-True([bool]$Condition, [string]$Message) {
+    $script:assertions++
+    if (-not $Condition) { throw $Message }
+}
+function Assert-Rejected([scriptblock]$Call) {
+    $before = $script:connections
+    $caught = $false
+    try { & $Call | Out-Null } catch {
+        $caught = $true
+        Assert-True ($_.Exception.Message -eq "unsupported_diagnostic_operation_stage_a") "Must reject before transport"
+    }
+    Assert-True $caught "Unsupported diagnostics must throw"
+    Assert-True ($script:connections -eq $before) "Rejected request must create zero sockets and write zero bytes"
+}
+function Import-Definitions([string]$Path) {
+    $tokens = $null; $errors = $null
+    $ast = [System.Management.Automation.Language.Parser]::ParseFile($Path, [ref]$tokens, [ref]$errors)
+    Assert-True ($errors.Count -eq 0) "Production script must parse"
+    foreach ($definition in $ast.FindAll({
+        param($node)
+        $node -is [System.Management.Automation.Language.FunctionDefinitionAst] -and
+        $node.Parent -is [System.Management.Automation.Language.NamedBlockAst]
+    }, $true)) {
+        . ([scriptblock]::Create("function global:" + $definition.Name + " " + $definition.Body.Extent.Text))
+    }
+}
+
+# Load only real function definitions: no config, Hub request, or scheduled loop.
+$root = Split-Path $PSScriptRoot -Parent
+Import-Definitions (Join-Path $root "EnrichOneMiner.ps1")
+Import-Definitions (Join-Path $root "MinerOpsApi3EnrichmentPoller.ps1")
+Import-Definitions (Join-Path $root "MinerOpsPoolPoller.ps1")
+
+function New-Object {
+    param([Parameter(Position=0)][string]$TypeName,
+          [Parameter(Position=1)][object[]]$ArgumentList)
+    if ($TypeName -eq "System.Net.Sockets.TcpClient") { $script:connections++ }
+    if ($null -ne $ArgumentList) {
+        return Microsoft.PowerShell.Utility\New-Object -TypeName $TypeName -ArgumentList $ArgumentList
+    }
+    return Microsoft.PowerShell.Utility\New-Object -TypeName $TypeName
+}
+
+# Auditor reproductions plus write families, alias/parameter overrides, injection,
+# malformed structures and operations that merely contain a read-looking substring.
+foreach ($command in @("reboot", "set.miner.service", "power_off", "switchpool", "ascset",
+    "summary|reboot", "summary`nreboot", 'summary","cmd":"reboot', "read_then_reboot",
+    "SUMMARY", "", "get.device.info", "get_miner_info|all")) {
+    Assert-Rejected { Invoke-MinerCommandRaw -IpAddress "127.0.0.1" -Command $command }
+}
+Assert-Rejected { Invoke-MinerCommandRaw -IpAddress "127.0.0.1" -Command "summary" -Port 4433 }
+$badRequests = @(
+    @{ cmd = "set.miner.service"; param = "stop" },
+    @{ cmd = "reboot" }, @{ cmd = "set.miner.power_mode"; param = "high" },
+    @{ cmd = "set.device.password"; param = "changed" }, @{ cmd = "get.read_then_write" },
+    @{ cmd = "get.device.info"; command = "reboot" },
+    @{ cmd = "get.device.info"; param = "stop" },
+    @{ cmd = "get.device.info"; account = "operator"; token = "anything" },
+    @{ cmd = "get.miner.status" }, @{ cmd = "get.miner.status"; param = "all" },
+    @{ cmd = "get.miner.status"; param = @("summary+pools+edevs") },
+    @{ cmd = "get.miner.status"; param = "summary+pools+edevs"; command = "reboot" },
+    @{ Cmd = "get.device.info" }, @{ cmd = "GET.DEVICE.INFO" },
+    @{ cmd = "get.device.info`nreboot" }, @{ cmd = @("get.device.info", "reboot") },
+    @{ cmd = $null }, @{}
+)
+foreach ($request in $badRequests) {
+    Assert-Rejected { Invoke-WhatsMinerApi3Read -Ip "127.0.0.1" -Request $request }
+    Assert-Rejected { Invoke-Api3OkCommand -Ip "127.0.0.1" -Request $request }
+}
+# Exact re-audit suffixes: U+0000, U+00AD and U+200B; also prefix/embedded
+# invisible characters, controls and whitespace. Never trim or normalize inputs.
+$originalCulture = [Threading.Thread]::CurrentThread.CurrentCulture
+try {
+    foreach ($cultureName in @("en-US", "tr-TR", "")) {
+        [Threading.Thread]::CurrentThread.CurrentCulture = [Globalization.CultureInfo]::GetCultureInfo($cultureName)
+        $variants = @()
+        foreach ($code in @(0, 0x00ad, 0x200b, 0x200c, 0x200d, 0x2060, 0xfeff,
+                             0x001f, 13, 10, 9, 32, 0x00a0)) {
+            $character = [string][char]$code
+            foreach ($position in @("prefix", "embedded", "suffix")) {
+                $variants += @{ Character = $character; Position = $position }
+            }
+        }
+        foreach ($variant in $variants) {
+            $legacy = "summary"; $api = "get.device.info"; $key = "cmd"; $param = "summary+pools+edevs"
+            if ($variant.Position -eq "prefix") {
+                $legacy = $variant.Character + $legacy; $api = $variant.Character + $api
+                $key = $variant.Character + $key; $param = $variant.Character + $param
+            } elseif ($variant.Position -eq "embedded") {
+                $legacy = "sum" + $variant.Character + "mary"; $api = "get.device." + $variant.Character + "info"
+                $key = "c" + $variant.Character + "md"; $param = "summary+" + $variant.Character + "pools+edevs"
+            } else {
+                $legacy += $variant.Character; $api += $variant.Character
+                $key += $variant.Character; $param += $variant.Character
+            }
+            Assert-Rejected { Invoke-MinerCommandRaw -IpAddress "127.0.0.1" -Command $legacy }
+            Assert-Rejected { Invoke-WhatsMinerApi3Read -Ip "127.0.0.1" -Request @{ cmd = $api } }
+            Assert-Rejected { Invoke-Api3OkCommand -Ip "127.0.0.1" -Request @{ cmd = $api } }
+            $request = @{}; $request[$key] = "get.device.info"
+            Assert-Rejected { Invoke-WhatsMinerApi3Read -Ip "127.0.0.1" -Request $request }
+            Assert-Rejected { Invoke-WhatsMinerApi3Read -Ip "127.0.0.1" -Request @{ cmd = "get.miner.status"; param = $param } }
+            $request = @{ cmd = "get.miner.status" }; $request[$key.Replace("cmd", "param")] = "summary+pools+edevs"
+            Assert-Rejected { Invoke-WhatsMinerApi3Read -Ip "127.0.0.1" -Request $request }
+        }
+        foreach ($legacy in @(("summar" + [char]0x0443), ([char]0xff53 + "ummary"),
+                              ("summ" + [char]0x0061 + [char]0x0301 + "ry"),
+                              ("summ" + [char]0x00e1 + "ry"), "Summary", "summary/", "summary+")) {
+            Assert-Rejected { Invoke-MinerCommandRaw -IpAddress "127.0.0.1" -Command $legacy }
+        }
+        foreach ($api in @(("get.dev" + [char]0x0456 + "ce.info"), ("get" + [char]0xff0e + "device.info"),
+                           ([char]0xff47 + "et.device.info"), "Get.Device.Info", "get.device.info/", "get.device.info+")) {
+            Assert-Rejected { Invoke-WhatsMinerApi3Read -Ip "127.0.0.1" -Request @{cmd=$api} }
+        }
+        Assert-Rejected { Invoke-WhatsMinerApi3Read -Ip "127.0.0.1" -Request @{cmd="get.miner.status"; Param="summary+pools+edevs"} }
+        Assert-Rejected { Invoke-WhatsMinerApi3Read -Ip "127.0.0.1" -Request @{cmd="get.miner.status"; param="Summary+pools+edevs"} }
+    }
+} finally { [Threading.Thread]::CurrentThread.CurrentCulture = $originalCulture }
+Assert-True ($script:connections -eq 0) "All negative cases produced zero transport connections/writes"
+
+# Real TCP compatibility tests: bind loopback only, one request per simulator.
+# This suite must be run with physical networks disconnected (Docker --network none).
+Add-Type -TypeDefinition @'
+using System;
+using System.Net;
+using System.Net.Sockets;
+using System.Text;
+using System.Threading;
+public class MinerOpsDiagnosticSimulator : IDisposable {
+    TcpListener listener;
+    Thread thread;
+    public string Payload;
+    public string Error;
+    public MinerOpsDiagnosticSimulator(int port, bool framed) {
+        listener = new TcpListener(IPAddress.Loopback, port); listener.Start();
+        thread = new Thread(() => {
+            try {
+                using (var client = listener.AcceptTcpClient()) {
+                    client.ReceiveTimeout = 2000;
+                    var stream = client.GetStream(); var buffer = new byte[65536]; int count;
+                    if (framed) {
+                        var length = new byte[4]; ReadExact(stream, length, 4);
+                        count = BitConverter.ToInt32(length, 0);
+                        if(count < 1 || count > buffer.Length) throw new Exception("Bad request length");
+                        ReadExact(stream, buffer, count);
+                    } else { count = stream.Read(buffer, 0, buffer.Length); }
+                    Payload = Encoding.ASCII.GetString(buffer, 0, count);
+                    var response = Encoding.ASCII.GetBytes("{\"code\":0,\"POOLS\":[],\"msg\":{}}");
+                    if(framed) { var length = BitConverter.GetBytes(response.Length); stream.Write(length,0,4); }
+                    stream.Write(response, 0, response.Length);
+                }
+            } catch(Exception e) { Error = e.Message; }
+            finally { listener.Stop(); }
+        }); thread.IsBackground = true; thread.Start();
+    }
+    static void ReadExact(NetworkStream stream, byte[] bytes, int count) {
+        int offset=0;
+        while(offset<count) { int n=stream.Read(bytes,offset,count-offset); if(n==0) throw new Exception("EOF"); offset+=n; }
+    }
+    public void Dispose() { listener.Stop(); thread.Join(3000); }
+}
+'@
+
+foreach ($command in @("summary", "devs", "edevs", "stats", "pools", "devdetails", "version", "get_version", "get_miner_info")) {
+    $miner = [MinerOpsDiagnosticSimulator]::new(4028, $false)
+    try {
+        $null = Invoke-MinerCommandRaw -IpAddress "127.0.0.1" -Command $command -TimeoutMs 100
+        $observed = $miner.Payload | ConvertFrom-Json
+        Assert-True ([string]::Equals($observed.cmd, $command, [StringComparison]::Ordinal)) "Legacy query must remain compatible"
+        Assert-True ($observed.PSObject.Properties.Count -eq 1) "Legacy request contains only selected read"
+        Assert-True (-not $miner.Error) "Simulator received a valid legacy request"
+    } finally { $miner.Dispose() }
+}
+foreach ($command in @("get.device.info", "get.miner.setting", "get.miner.status", "get.fan.setting", "get.device.custom_data")) {
+    $request = @{ cmd = $command }
+    if ($command -ceq "get.miner.status") { $request.param = "summary+pools+edevs" }
+    $miner = [MinerOpsDiagnosticSimulator]::new(4433, $true)
+    try {
+        $result = Invoke-Api3OkCommand -Ip "127.0.0.1" -Request $request
+        $observed = $miner.Payload | ConvertFrom-Json
+        Assert-True ($null -ne $result) "Existing API3 reader wrapper remains compatible"
+        Assert-True ([string]::Equals($observed.cmd, $command, [StringComparison]::Ordinal)) "Exact API3 query transmitted"
+        Assert-True (@($observed.PSObject.Properties).Count -eq $request.Count) "No additional API3 request keys"
+        if ($command -ceq "get.miner.status") {
+            Assert-True ([string]::Equals($observed.param, "summary+pools+edevs", [StringComparison]::Ordinal)) "Only exact read parameter transmitted"
+        }
+        Assert-True (-not $miner.Error) "Simulator received a valid framed read"
+    } finally { $miner.Dispose() }
+}
+$miner = [MinerOpsDiagnosticSimulator]::new(4028, $false)
+try {
+    $null = Invoke-MinerOps4028PoolsSimple -Ip "127.0.0.1"
+    Assert-True (($miner.Payload | ConvertFrom-Json).command -ceq "pools") "Pool poller sends a constant read"
+} finally { $miner.Dispose() }
+Write-Host "Diagnostic containment passed: $($script:assertions) assertions; all rejected operations made zero connections/writes."

agents/windows/tests/Test-MinerOpsOldAgentFence.ps1

@@ -0,0 +1,46 @@
+param(
+    [Parameter(Mandatory=$true)][string]$BaselineRoot,
+    [Parameter(Mandatory=$true)][string]$CachedPayloadPath,
+    [Parameter(Mandatory=$true)][string]$MinerIp,
+    [switch]$ExpectFenced
+)
+$ErrorActionPreference = "Stop"
+if ($env:MINEROPS_STAGE_A_SIMULATION -ne "1") {
+    throw "Requires an explicitly isolated Stage A simulation; never use physical targets."
+}
+
+# Supply a git-archived 2d89e80 Windows source directory, NOT a running installation.
+# Only AST function definitions execute; no startup, config loading, or task loops.
+$tokens = $null; $errors = $null
+$ast = [System.Management.Automation.Language.Parser]::ParseFile(
+    (Join-Path $BaselineRoot "MinerOpsAgentService.ps1"), [ref]$tokens, [ref]$errors)
+if ($errors.Count) { throw "Baseline parser error" }
+foreach ($definition in $ast.FindAll({
+    param($node)
+    $node -is [System.Management.Automation.Language.FunctionDefinitionAst] -and
+    $node.Parent -is [System.Management.Automation.Language.NamedBlockAst]
+}, $true)) {
+    Invoke-Expression $definition.Extent.Text
+}
+. (Join-Path $BaselineRoot "MinerOpsControl.ps1")
+function Read-AgentConfig {
+    return [pscustomobject]@{
+        api3_account = "simulation"; api3_password = "simulation-only"; api3_port = 4433
+    }
+}
+function Get-MinerOpsApi3Salt { return @{ salt = "simulation-only"; salt_response = @{} } }
+
+$payload = Get-Content -LiteralPath $CachedPayloadPath -Raw | ConvertFrom-Json
+if ($payload.action -ne "suspend") { throw "This simulation expects the audited cached suspend command" }
+# Remap ONLY the target to the address of the dedicated disposable miner container.
+$payload.ip_address = $MinerIp
+$result = Run-MinerControlCommand -Payload $payload
+if ($ExpectFenced) {
+    if ($result.command_accepted -ne $false -or $result.failure_code -ne "write_rejected") {
+        throw "Old executor unexpectedly reached/accepted a simulated miner through the fence"
+    }
+    Write-Host "Fenced old cached command failed; independently verify simulator write count stayed zero."
+} else {
+    if ($result.command_accepted -ne $true) { throw "Unfenced control simulation did not reproduce old execution" }
+    Write-Host "Unfenced old cached command reproduced; simulated miner received service stop."
+}

backend/app/agent_commands/policy.py

@@ -0,0 +1,35 @@
+"""Stage A containment: no current payload authorizes a physical write.
+
+There is deliberately no environment-variable override. Re-enabling physical
+control requires a separately reviewed identity-binding protocol.
+"""
+
+PHYSICAL_CONTROL_DISABLED = "physical_control_disabled_stage_a"
+UNSUPPORTED_COMMAND = "unsupported_command_schema_stage_a"
+READ_ONLY_COMMAND_TYPES = (
+    "poll_known_miners_now",
+    "discover_new_miners",
+    "enrich_match_discovery",
+)
+
+
+def rejection_reason(command_type: str) -> str | None:
+    if command_type == "miner_control":
+        return PHYSICAL_CONTROL_DISABLED
+    if command_type not in READ_ONLY_COMMAND_TYPES:
+        return UNSUPPORTED_COMMAND
+    return None
+
+
+def rejection_execution_outcome(row: dict) -> str:
+    """A current pending state alone says nothing about earlier delivery.
+
+    The repository supplies an explicit pre-transition evidence check while
+    holding the queue row lock. Missing/ambiguous checks and result submissions
+    are unknown. This describes queue delivery, never proof of physical outcome.
+    """
+    if (row.get("previous_status") == "pending"
+            and row.get("no_delivery_evidence") is True
+            and row.get("delivery_reported", False) is False):
+        return "not_dispatched"
+    return "unknown"

backend/app/agent_commands/repository.py

@@ -8,6 +8,7 @@ from psycopg.rows import dict_row
 from psycopg.types.json import Jsonb
 
 from app.agent_commands.exceptions import AgentCommandConflict, AgentCommandNotFound
+from app.agent_commands.policy import READ_ONLY_COMMAND_TYPES, rejection_execution_outcome, rejection_reason
 from app.config import settings
 
 
@@ -19,6 +20,7 @@ WITH candidates AS (
     FROM minerops_agent_command_queue
     WHERE agent_id = %s
       AND command_status = 'pending'
+      AND command_type = ANY(%s)
     ORDER BY created_at, id
     FOR UPDATE SKIP LOCKED
     LIMIT %s
@@ -32,6 +34,70 @@ RETURNING queue.id AS command_id, queue.command_type,
           queue.request_payload, queue.created_at
 """
 
+# Inspect existing durable history before changing status/completed_at/error.
+# A linked non-creation audit is conservatively ambiguous, even if timestamps
+# were lost. Legacy results need not agree on Agent ID to prevent a false claim
+# of non-delivery. Missing tables/permissions fail the transaction closed.
+NO_DELIVERY_EVIDENCE_SQL = """
+    q.created_at IS NOT NULL
+    AND q.claimed_at IS NULL AND q.completed_at IS NULL
+    AND q.result_payload IS NULL AND COALESCE(q.error_message, '') = ''
+    AND NOT EXISTS (
+        SELECT 1 FROM minerops_agent_command_results r
+        WHERE r.command_id = q.id
+           OR (r.agent_id = q.agent_id AND NULLIF(btrim(r.command_id), '') IS NULL)
+    )
+    AND NOT EXISTS (
+        SELECT 1 FROM audit_logs a
+        WHERE a.action <> 'miner.control.queued'
+          AND (a.old_value_json ->> 'command_id' = q.id
+               OR a.new_value_json ->> 'command_id' = q.id
+               OR a.old_value_json ->> 'operation_id' = q.id
+               OR a.new_value_json ->> 'operation_id' = q.id
+               OR (a.entity_type = 'agent_command' AND a.entity_id::text = q.id))
+    )
+"""
+
+RETIRE_COMMANDS_SQL = f"""
+WITH candidates AS (
+    SELECT q.id, q.command_status, ({NO_DELIVERY_EVIDENCE_SQL}) AS no_delivery_evidence
+    FROM minerops_agent_command_queue q
+    WHERE agent_id = %s AND command_status IN ('pending', 'claimed')
+      AND NOT (command_type = ANY(%s))
+    ORDER BY created_at, id
+    FOR UPDATE SKIP LOCKED
+    LIMIT 1000
+)
+UPDATE minerops_agent_command_queue AS queue
+SET command_status = 'cancelled', completed_at = COALESCE(queue.completed_at, now()),
+    error_message = concat_ws(chr(10), NULLIF(queue.error_message, ''), CASE WHEN queue.command_type = 'miner_control'
+        THEN 'physical_control_disabled_stage_a'
+        ELSE 'unsupported_command_schema_stage_a' END)
+FROM candidates WHERE queue.id = candidates.id
+RETURNING queue.id, queue.agent_id, queue.command_type,
+          candidates.command_status AS previous_status, candidates.no_delivery_evidence, queue.error_message
+"""
+
+RETIRE_SCHEDULE_COMMANDS_SQL = f"""
+WITH candidates AS (
+    SELECT q.id, q.command_status, ({NO_DELIVERY_EVIDENCE_SQL}) AS no_delivery_evidence
+    FROM minerops_agent_command_queue q
+    WHERE agent_id = %s AND request_payload ->> 'scheduled_action_id' = %s
+      AND command_status IN ('pending', 'claimed')
+      AND NOT (command_type = ANY(%s))
+    ORDER BY id
+    FOR UPDATE
+)
+UPDATE minerops_agent_command_queue AS queue
+SET command_status = 'cancelled', completed_at = COALESCE(queue.completed_at, now()),
+    error_message = concat_ws(chr(10), NULLIF(queue.error_message, ''), CASE WHEN queue.command_type = 'miner_control'
+        THEN 'physical_control_disabled_stage_a'
+        ELSE 'unsupported_command_schema_stage_a' END)
+FROM candidates WHERE queue.id = candidates.id
+RETURNING queue.id, queue.agent_id, queue.command_type,
+          candidates.command_status AS previous_status, candidates.no_delivery_evidence
+"""
+
 
 def default_connection_factory() -> psycopg.Connection:
     return psycopg.connect(
@@ -54,7 +120,8 @@ class AgentCommandRepository:
         connection = self._connection_factory()
         try:
             with connection.cursor(row_factory=dict_row) as cur:
-                cur.execute(CLAIM_COMMANDS_SQL, (agent_id, limit))
+                self._retire_unsupported(cur, agent_id)
+                cur.execute(CLAIM_COMMANDS_SQL, (agent_id, list(READ_ONLY_COMMAND_TYPES), limit))
                 rows = [dict(row) for row in cur.fetchall()]
             connection.commit()
             return rows
@@ -64,6 +131,105 @@ class AgentCommandRepository:
         finally:
             connection.close()
 
+    @staticmethod
+    def _audit_rejection(
+        cur, row: dict, *, actor_user_id: str | None = None,
+        source: str = "agent_poll",
+    ) -> None:
+        # Queue IDs can be legacy text, not UUIDs. Preserve them in JSON rather
+        # than coercing them into audit_logs.entity_id. Never log raw payloads.
+        cur.execute(
+            """INSERT INTO audit_logs
+               (actor_user_id, action, entity_type, old_value_json, new_value_json)
+               VALUES (%s, 'agent.command.rejected', 'agent_command', %s, %s)""",
+            (actor_user_id, Jsonb({"command_id": row["id"], "agent_id": row["agent_id"],
+                    "status": row["previous_status"]}),
+             Jsonb({"status": "cancelled", "reason": rejection_reason(row["command_type"]),
+                    "source": source,
+                    "execution_outcome": rejection_execution_outcome(row)})),
+        )
+
+    @classmethod
+    def _retire_unsupported(
+        cls, cur, agent_id: str, *, source: str = "agent_poll",
+        actor_user_id: str | None = None,
+    ) -> int:
+        cur.execute(RETIRE_COMMANDS_SQL, (agent_id, list(READ_ONLY_COMMAND_TYPES)))
+        rows = list(cur.fetchall())
+        for row in rows:
+            cls._audit_rejection(cur, row, source=source, actor_user_id=actor_user_id)
+        return len(rows)
+
+    def retire_unsupported(self, agent_id: str, *, actor_user_id: str) -> int:
+        """Audited cutover batch, including offline Agents; never claim any work.
+
+        A zero batch can mean locked rows were skipped. Operators must separately
+        verify no unsupported live rows remain before declaring a drain complete.
+        """
+        if not actor_user_id:
+            raise ValueError("Cutover retirement requires an accountable operator actor")
+        connection = self._connection_factory()
+        try:
+            with connection.cursor(row_factory=dict_row) as cur:
+                count = self._retire_unsupported(cur, agent_id, source="stage_a_cutover",
+                                                 actor_user_id=actor_user_id)
+            connection.commit()
+            return count
+        except Exception:
+            connection.rollback()
+            raise
+        finally:
+            connection.close()
+
+    def retire_schedule(self, schedule_id: str, *, actor_user_id: str, operation: str) -> int:
+        """Cancel live unsupported work and retain schedule history atomically.
+
+        The stored schedule determines Agent scope. Delete retires the schedule;
+        neither operation erases evidence or changes terminal command history.
+        """
+        if not actor_user_id or operation not in {"cancel", "delete"}:
+            raise ValueError("An authenticated actor and cancel/delete operation are required")
+        connection = self._connection_factory()
+        try:
+            with connection.cursor(row_factory=dict_row) as cur:
+                # Serialize retries and cancel/delete workers for this schedule.
+                cur.execute(
+                    """SELECT id, agent_id, status FROM minerops_scheduled_bulk_actions
+                       WHERE id = %s FOR UPDATE""", (schedule_id,),
+                )
+                schedule = cur.fetchone()
+                if not schedule:
+                    raise AgentCommandNotFound("Scheduled action not found")
+                cur.execute(RETIRE_SCHEDULE_COMMANDS_SQL,
+                            (schedule["agent_id"], schedule_id, list(READ_ONLY_COMMAND_TYPES)))
+                rows = list(cur.fetchall())
+                for row in rows:
+                    self._audit_rejection(cur, row, actor_user_id=actor_user_id,
+                                          source=f"schedule_{operation}")
+                if schedule["status"] in {"pending", "queued", "running"}:
+                    cur.execute(
+                        """UPDATE minerops_scheduled_bulk_actions
+                           SET status = 'cancelled', updated_at = now(),
+                               last_error = concat_ws(chr(10), NULLIF(last_error, ''),
+                                                      'physical_control_disabled_stage_a')
+                           WHERE id = %s""", (schedule_id,),
+                    )
+                    cur.execute(
+                        """INSERT INTO audit_logs
+                           (actor_user_id, action, entity_type, old_value_json, new_value_json)
+                           VALUES (%s, 'scheduled_action.retired', 'scheduled_action', %s, %s)""",
+                        (actor_user_id, Jsonb(dict(schedule)),
+                         Jsonb({"status": "cancelled", "operation": operation,
+                                "reason": "physical_control_disabled_stage_a"})),
+                    )
+            connection.commit()
+            return len(rows)
+        except Exception:
+            connection.rollback()
+            raise
+        finally:
+            connection.close()
+
     def get_command(self, command_id: str) -> dict | None:
         connection = self._connection_factory()
         try:
@@ -91,6 +257,32 @@ class AgentCommandRepository:
         connection = self._connection_factory()
         try:
             with connection.cursor(row_factory=dict_row) as cur:
+                # Enforce the same boundary on results: an old claimed entry
+                # cannot be promoted by an old Agent, retry, or direct caller.
+                cur.execute(
+                    f"""SELECT q.id, q.agent_id, q.command_type, q.command_status AS previous_status,
+                              ({NO_DELIVERY_EVIDENCE_SQL}) AS no_delivery_evidence
+                       FROM minerops_agent_command_queue q
+                       WHERE q.id = %s AND q.agent_id = %s FOR UPDATE""",
+                    (command_id, agent_id),
+                )
+                existing = cur.fetchone()
+                if existing and rejection_reason(existing["command_type"]):
+                    reason = rejection_reason(existing["command_type"])
+                    if existing["previous_status"] in {"pending", "claimed"}:
+                        cur.execute(
+                            """UPDATE minerops_agent_command_queue
+                               SET command_status = 'cancelled',
+                                   error_message = concat_ws(chr(10), NULLIF(error_message, ''), %s::text),
+                                   completed_at = COALESCE(completed_at, now()) WHERE id = %s""",
+                            (reason, command_id),
+                        )
+                        # A result submission itself reports possible delivery, even
+                        # when earlier durable claim metadata is absent.
+                        existing["delivery_reported"] = True
+                        self._audit_rejection(cur, existing, source="agent_result")
+                    connection.commit()
+                    raise AgentCommandConflict(reason)
                 cur.execute(
                     """
                     UPDATE minerops_agent_command_queue

backend/app/agent_commands/router.py

@@ -1,6 +1,4 @@
 """HTTP boundary for authenticated Agent command transport."""
-
-import logging
 from typing import Annotated
 
 from fastapi import APIRouter, Depends, HTTPException, Request
@@ -18,8 +16,6 @@ from app.agent_commands.models import (
 )
 from app.agent_commands.service import AgentCommandService
 
-
-logger = logging.getLogger(__name__)
 router = APIRouter(prefix="/api/agent/commands", tags=["agent-commands"])
 AgentIdentity = Annotated[str, Depends(require_agent_identity)]
 AgentService = Annotated[AgentCommandService, Depends(get_agent_command_service)]
@@ -29,14 +25,7 @@ AgentService = Annotated[AgentCommandService, Depends(get_agent_command_service)
 def claim_pending_commands(
     request: Request, agent_id: AgentIdentity, service: AgentService
 ) -> AgentCommandListResponse:
-    before_claim = getattr(
-        request.app.state, "agent_command_before_claim", None
-    )
-    if callable(before_claim):
-        try:
-            before_claim()
-        except Exception:
-            logger.exception("Scheduled command materialization failed before Agent claim")
+    del request
     return AgentCommandListResponse(commands=service.claim(agent_id))
 
 

backend/app/agent_commands/service.py

@@ -2,19 +2,16 @@
 
 import re
 
-from pydantic import ValidationError
-
 from app.agent_commands.exceptions import (
     AgentCommandConflict,
     AgentCommandNotFound,
-    AgentResultValidationError,
 )
 from app.agent_commands.models import (
     AgentCommand,
     AgentCommandResultRequest,
-    MinerControlResultPayload,
 )
 from app.agent_commands.repository import AgentCommandRepository
+from app.agent_commands.policy import rejection_reason
 
 
 class AgentCommandService:
@@ -25,6 +22,7 @@ class AgentCommandService:
         return [
             AgentCommand.model_validate(row)
             for row in self.repository.claim_pending(agent_id)
+            if rejection_reason(row["command_type"]) is None
         ]
 
     @staticmethod
@@ -42,6 +40,14 @@ class AgentCommandService:
             raise AgentCommandNotFound("Agent command not found")
         if command["agent_id"] != agent_id:
             raise AgentCommandConflict("Agent command belongs to another Agent")
+        reason = rejection_reason(command["command_type"])
+        if reason:
+            # Repository repeats this check under a row lock and records the
+            # cancellation atomically; never accept an old physical result.
+            self.repository.finish_command(
+                request.command_id, agent_id, "cancelled", {}, reason
+            )
+            raise AgentCommandConflict(reason)
         if command["command_status"] != "claimed":
             raise AgentCommandConflict(
                 f"Agent command is {command['command_status']}, not claimed"
@@ -49,34 +55,6 @@ class AgentCommandService:
 
         result_payload = request.result_payload
         error_message = self._legacy_error(request.error_message)
-        if command["command_type"] == "miner_control":
-            try:
-                result = MinerControlResultPayload.model_validate(request.result_payload)
-            except ValidationError as exc:
-                raise AgentResultValidationError(
-                    "Invalid typed miner-control result"
-                ) from exc
-            if request.status == "completed" and not result.command_accepted:
-                raise AgentResultValidationError(
-                    "Completed miner-control results must be accepted"
-                )
-            if request.status == "failed" and result.command_accepted:
-                raise AgentResultValidationError(
-                    "Failed miner-control results cannot be accepted"
-                )
-            action = (command.get("request_payload") or {}).get("action")
-            if (
-                action in {"suspend", "resume"}
-                and result.command_accepted
-                and result.verification_status != "unconfirmed"
-            ):
-                raise AgentResultValidationError(
-                    "Accepted suspend and resume results must remain unconfirmed"
-                )
-            result_payload = result.model_dump(mode="json")
-            error_message = (
-                result.failure_code.replace("_", " ") if result.failure_code else None
-            )
 
         self.repository.finish_command(
             request.command_id,

backend/app/identity.py

@@ -708,6 +708,8 @@ def machine_agent_request(path: str) -> bool:
         "/api/agent/pool-snapshots",
         "/api/agent/commands/pending",
         "/api/agent/commands/result",
+        "/api/agent/commands/pending-disabled-old",
+        "/api/agent/commands/result-disabled-old",
         "/api/agent/discoveries",
         "/api/agent/miner-enrichment",
         "/api/agent/miner-enrichment-v2",
@@ -760,6 +762,19 @@ def component_request(path: str) -> bool:
 
 def required_api_permission(method: str, path: str) -> str | None:
     parts = path.split("/")
+    if method == "POST" and path == "/api/dashboard/m30-command-test-lab/run":
+        return "miners.control"
+    if method == "POST" and path == "/api/tools/whatsminer-read-test":
+        return "miners.control"
+    if method == "POST" and path in {
+        "/api/dashboard/scheduled-bulk-actions",
+        "/api/dashboard/scheduled-bulk-actions/process-due",
+    }:
+        return "miners.control"
+    if (method == "POST" and len(parts) == 6
+            and parts[:4] == ["", "api", "dashboard", "scheduled-bulk-actions"]
+            and parts[5] in {"cancel", "delete"}):
+        return "miners.control"
     if (
         len(parts) >= 5
         and parts[:3] == ["", "api", "miners"]

backend/app/miner_control/service.py

@@ -1,7 +1,6 @@
 """Business rules for typed, Agent-routed miner controls."""
 
-from typing import Any
-from uuid import UUID, uuid4
+from uuid import UUID
 
 from app.miner_control.exceptions import (
     ControlledMinerNotFound,
@@ -16,6 +15,7 @@ from app.miner_control.models import (
     MinerControlProjection,
 )
 from app.miner_control.repository import MinerControlRepository
+from app.agent_commands.policy import PHYSICAL_CONTROL_DISABLED
 
 
 KNOWN_MODES = {"low", "normal", "high"}
@@ -29,33 +29,8 @@ class MinerControlService:
 
     @staticmethod
     def _capability(context: dict) -> ControlCapability:
-        if not context.get("agent_zone_id"):
-            return ControlCapability(state="unavailable", reason="agent_zone_unassigned")
-        if context.get("agent_zone_exists") is not True:
-            return ControlCapability(state="unavailable", reason="agent_zone_missing")
-        if context.get("agent_zone_enabled") is not True:
-            return ControlCapability(state="unavailable", reason="agent_zone_disabled")
-        if not context.get("agent_id"):
-            return ControlCapability(state="unavailable", reason="agent_unavailable")
-        if not context.get("current_ip"):
-            return ControlCapability(state="unavailable", reason="current_ip_missing")
-        if context.get("api3_supported") is False:
-            return ControlCapability(state="unavailable", reason="api3_unsupported")
-        if context.get("api3_supported") is not True:
-            return ControlCapability(state="unknown", reason="api3_support_unknown")
-
-        payload = context.get("last_payload") or {}
-        if payload.get("control_available") is False:
-            reason = payload.get("control_unavailable_reason")
-            return ControlCapability(
-                state="unavailable",
-                reason=reason if reason == "credentials_missing" else "agent_unavailable",
-            )
-        if payload.get("control_available") is not True:
-            return ControlCapability(
-                state="unknown", reason="agent_capability_unknown"
-            )
-        return ControlCapability(state="supported")
+        # Credentials and Zone routing are not physical identity authorization.
+        return ControlCapability(state="unavailable", reason=PHYSICAL_CONTROL_DISABLED)
 
     @staticmethod
     def _operation(miner_id: UUID, row: dict) -> MinerControlOperation:
@@ -120,50 +95,4 @@ class MinerControlService:
     ) -> MinerControlAccepted:
         if not actor_user_id:
             raise MinerControlActorRequired("Authenticated actor UUID is required")
-        context = self.repository.get_control_context(miner_id)
-        if not context:
-            raise ControlledMinerNotFound("Miner not found")
-        capability = self._capability(context)
-        if capability.state != "supported":
-            raise MinerControlUnavailable(
-                capability.reason or "control_capability_not_supported"
-            )
-
-        runtime = str(context.get("runtime_state") or "").strip().lower()
-        if action == "suspend" and runtime not in RUNNING_STATES:
-            raise MinerControlUnavailable("miner_not_factually_running")
-        if action == "resume" and runtime not in STOPPED_STATES:
-            raise MinerControlUnavailable("miner_not_factually_stopped")
-        if action == "performance_mode" and requested_value not in KNOWN_MODES:
-            raise MinerControlUnavailable("invalid_performance_mode")
-
-        operation_id = str(uuid4())
-        payload: dict[str, Any] = {
-            "schema_version": 1,
-            "miner_id": str(miner_id),
-            "action": action,
-            "requested_value": requested_value,
-            "ip_address": context["current_ip"],
-        }
-        operation = self.repository.enqueue_with_audit(
-            operation_id=operation_id,
-            miner_id=miner_id,
-            agent_id=context["agent_id"],
-            actor_user_id=actor_user_id,
-            actor_username=actor_username,
-            request_payload=payload,
-            audit_payload={
-                "operation_id": operation_id,
-                "miner_id": str(miner_id),
-                "action": action,
-                "requested_value": requested_value,
-            },
-        )
-        return MinerControlAccepted(
-            operation_id=operation["operation_id"],
-            miner_id=miner_id,
-            action=action,
-            requested_value=requested_value,
-            status=operation["status"],
-            created_at=operation["created_at"],
-        )
+        raise MinerControlUnavailable(PHYSICAL_CONTROL_DISABLED)

backend/tests/test_agent_commands.py

@@ -16,7 +16,7 @@ from app.agent_commands.exceptions import (
     AgentCommandConflict,
     AgentResultValidationError,
 )
-from app.agent_commands.models import AgentCommandResultRequest
+from app.agent_commands.models import AgentCommandResultRequest, MinerControlResultPayload
 from app.agent_commands.repository import CLAIM_COMMANDS_SQL
 from app.agent_commands.router import router
 from app.agent_commands.service import AgentCommandService
@@ -38,14 +38,14 @@ class FakeRepository:
     def __init__(self, row=None):
         self.row = row or {
             "id": "command-1", "agent_id": "test-agent",
-            "command_type": "miner_control", "command_status": "claimed",
+            "command_type": "poll_known_miners_now", "command_status": "claimed",
             "request_payload": {"action": "suspend"},
         }
         self.finished = []
 
     def claim_pending(self, agent_id, limit=5):
         return [{
-            "command_id": "command-1", "command_type": "miner_control",
+            "command_id": "command-1", "command_type": "poll_known_miners_now",
             "request_payload": {"action": "suspend"}, "created_at": NOW,
         }]
 
@@ -54,6 +54,8 @@ class FakeRepository:
 
     def finish_command(self, *args):
         self.finished.append(args)
+        if args[2] == "cancelled":
+            raise AgentCommandConflict(args[4])
 
 
 def typed_result(**overrides):
@@ -84,7 +86,7 @@ class AgentCommandServiceTests(unittest.TestCase):
         )
         AgentCommandService(repository).submit_result("test-agent", request)
         finished = repository.finished[0]
-        self.assertIsNone(finished[4])
+        self.assertIn("[address]", finished[4])
         self.assertNotIn("192.0.2.10", str(finished))
 
     def test_wrong_agent_pending_and_terminal_results_are_rejected(self):
@@ -111,8 +113,8 @@ class AgentCommandServiceTests(unittest.TestCase):
                     command_id="command-1", status="completed",
                     result_payload={**typed_result(), forbidden: "sensitive"},
                 )
-                with self.assertRaises(AgentResultValidationError):
-                    AgentCommandService(repository).submit_result("test-agent", request)
+                with self.assertRaises(ValueError):
+                    MinerControlResultPayload.model_validate(request.result_payload)
 
     def test_suspend_resume_cannot_claim_confirmation(self):
         for status in ("confirmed", "verification_failed"):
@@ -123,8 +125,10 @@ class AgentCommandServiceTests(unittest.TestCase):
                         verification_attempted=True, verification_status=status
                     ),
                 )
-                with self.assertRaises(AgentResultValidationError):
-                    AgentCommandService(FakeRepository()).submit_result("test-agent", request)
+                repository = FakeRepository()
+                repository.row["command_type"] = "miner_control"
+                with self.assertRaisesRegex(AgentCommandConflict, "physical_control_disabled_stage_a"):
+                    AgentCommandService(repository).submit_result("test-agent", request)
 
 
 class FakeAgentAuthRepository:
@@ -160,17 +164,26 @@ class AgentCommandRouterTests(unittest.TestCase):
             "X-MinerOps-Agent-Secret": "test-secret",
         }
 
-    def test_poll_requires_header_identity_and_ignores_query_spoofing(self):
+    def test_poll_accepts_secret_only_legacy_agent_and_ignores_query_spoofing(self):
         before_claim_calls = []
         self.app.state.agent_command_before_claim = lambda: before_claim_calls.append(True)
         self.assertEqual(self.client.get("/api/agent/commands/pending").status_code, 401)
+        self.assertEqual(
+            self.client.get(
+                "/api/agent/commands/pending",
+                headers={"X-MinerOps-Agent-Secret": "test-secret"},
+            ).status_code,
+            200,
+        )
         self.assertEqual(
             self.client.get(
                 "/api/agent/commands/pending?agent_id=other", headers=self.headers()
             ).status_code,
             200,
         )
-        self.assertEqual(before_claim_calls, [True])
+        # Agent polling may claim already queued commands, but must never invoke
+        # a legacy scheduler/materializer as a side effect.
+        self.assertEqual(before_claim_calls, [])
         self.assertEqual(
             self.client.get(
                 "/api/agent/commands/pending", headers=self.headers("other")
@@ -214,9 +227,9 @@ class AgentSourceContractTests(unittest.TestCase):
         self.assertIn('.agent_zone_id', service)
         self.assertIn('"agent_zone_id"', (root / "config.example.json").read_text())
         self.assertNotIn("?agent_id=", service)
-        self.assertIn('"get.miner.setting"', control)
-        self.assertIn("$attempt -le 3", control)
-        self.assertIn("$Payload.performance_mode", control)
+        self.assertIn("physical_control_disabled_stage_a", control)
+        self.assertNotIn('"set.miner.', control)
+        self.assertNotIn("$Payload.performance_mode", control)
         self.assertNotIn("api3_account =", control)
         self.assertNotIn("api3_password =", control)
         for forbidden in ("raw_response", "raw_command", "raw_exception"):
@@ -366,3 +379,18 @@ class AgentEnrichmentIsolationSourceTests(unittest.TestCase):
         self.assertIn("minerops_require_agent_secret(request)", route)
         self.assertIn('"site_name": "Plexus"', route)
         self.assertNotIn("Depends(require_agent_identity)", route)
+
+    def test_disabled_old_command_routes_are_fixed_retirement_responses(self):
+        main = (Path(__file__).parents[1] / "app" / "main.py").read_text()
+        for decorator in (
+            '@app.get("/api/agent/commands/pending-disabled-old")',
+            '@app.post("/api/agent/commands/result-disabled-old")',
+        ):
+            with self.subTest(decorator=decorator):
+                start = main.index(decorator)
+                end = main.index("\n@app.", start + 1)
+                route = main[start:end]
+                self.assertIn("status_code=410", route)
+                self.assertNotIn("minerops_require_agent_secret", route)
+                self.assertNotIn("request.json", route)
+                self.assertNotIn("minerops_agent_command_queue", route)

backend/tests/test_agent_zones_postgres.py

@@ -2,6 +2,7 @@
 
 import unittest
 from uuid import UUID
+from unittest.mock import patch
 
 import psycopg
 from psycopg.errors import ForeignKeyViolation, UniqueViolation
@@ -10,6 +11,8 @@ from app.agent_zones.repository import AgentZoneRepository
 from app.config import settings
 from app.miner_control.repository import MinerControlRepository
 from app.miner_control.service import MinerControlService
+from app.miner_control.exceptions import MinerControlUnavailable
+from app.agent_commands.policy import PHYSICAL_CONTROL_DISABLED
 
 
 SITE_A = UUID("00000000-0000-4000-8000-000000001630")
@@ -34,6 +37,29 @@ def connect():
 
 
 class AgentZonePostgresTests(unittest.TestCase):
+    def assert_stage_a_unavailable(self, expected_zone, expected_agent):
+        repository = MinerControlRepository(connect)
+        context = repository.get_control_context(MINER_UNASSIGNED)
+        # Check actual routing facts independently of the unconditional Stage A
+        # reason. A changed assertion must not hide a Site fallback regression.
+        self.assertEqual(context["agent_zone_id"], expected_zone)
+        self.assertEqual(context["agent_id"], expected_agent)
+        projection = MinerControlService(repository).get_controls(MINER_UNASSIGNED)
+        for capability in (projection.service_control, projection.performance_mode_control):
+            self.assertEqual(capability.state, "unavailable")
+            self.assertEqual(capability.reason, PHYSICAL_CONTROL_DISABLED)
+        self.assertIsNone(projection.available_service_action)
+        with patch.object(repository, "enqueue_with_audit") as enqueue:
+            for action, value in (("suspend", None), ("resume", None), ("performance_mode", "high")):
+                with self.assertRaisesRegex(MinerControlUnavailable, PHYSICAL_CONTROL_DISABLED):
+                    MinerControlService(repository).queue_control(
+                        MINER_UNASSIGNED, action, requested_value=value,
+                        actor_user_id=UUID("00000000-0000-4000-8000-000000001639"),
+                        actor_username="zone-containment-test",
+                    )
+            enqueue.assert_not_called()
+        return context
+
     @classmethod
     def setUpClass(cls):
         connection = connect()
@@ -203,10 +229,8 @@ class AgentZonePostgresTests(unittest.TestCase):
         finally:
             connection.close()
         try:
-            projection = MinerControlService(MinerControlRepository(connect)).get_controls(
-                MINER_UNASSIGNED
-            )
-            self.assertEqual(projection.service_control.reason, "agent_zone_disabled")
+            context = self.assert_stage_a_unavailable(ZONE_A, AGENT_A)
+            self.assertFalse(context["agent_zone_enabled"])
         finally:
             connection = connect()
             try:
@@ -228,16 +252,10 @@ class AgentZonePostgresTests(unittest.TestCase):
             connection.commit()
         finally:
             connection.close()
-        projection = MinerControlService(MinerControlRepository(connect)).get_controls(
-            MINER_UNASSIGNED
-        )
-        self.assertEqual(projection.service_control.reason, "agent_unavailable")
+        self.assert_stage_a_unavailable(ZONE_C, None)
 
     def test_null_zone_does_not_fall_back_to_site_agent(self):
-        projection = MinerControlService(MinerControlRepository(connect)).get_controls(
-            MINER_UNASSIGNED
-        )
-        self.assertEqual(projection.service_control.reason, "agent_zone_unassigned")
+        self.assert_stage_a_unavailable(None, None)
 
 
 if __name__ == "__main__":

backend/tests/test_legacy_control_retirement.py

@@ -0,0 +1,243 @@
+"""Focused regression tests for retired legacy miner-write surfaces."""
+
+import asyncio
+import inspect
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+from fastapi import FastAPI, HTTPException
+from fastapi.testclient import TestClient
+
+from app import main
+from app.identity import get_current_user, require_permission, required_api_permission
+
+
+APP_ROOT = Path(__file__).parents[1] / "app"
+
+
+class LegacyControlRetirementTests(unittest.TestCase):
+    @staticmethod
+    def assert_retired(callable_, *args, **kwargs):
+        try:
+            result = callable_(*args, **kwargs)
+            if inspect.isawaitable(result):
+                asyncio.run(result)
+        except HTTPException as exc:
+            if exc.status_code != 410:
+                raise AssertionError(f"Expected HTTP 410, got {exc.status_code}") from exc
+            return
+        raise AssertionError("Expected a fixed HTTP 410 retirement response")
+
+    def test_all_legacy_http_write_routes_are_fixed_inert_responses(self):
+        routes = (
+            (main.run_dashboard_m30_command_test_lab, {"role": "Admin"}),
+            (main.retired_direct_reboot_miner_command, "miner-1", {"role": "Admin"}),
+            (main.real_power_off_miner_command, "miner-1", {"role": "Admin"}),
+            (main.real_power_on_miner_command, "miner-1", {"role": "Admin"}),
+            (main.real_suspend_miner_command, "miner-1", {"role": "Admin"}),
+            (main.real_resume_miner_command, "miner-1", {"role": "Admin"}),
+            (main.retired_direct_api3_password_change, "miner-1", {"role": "Admin"}),
+            (main.retired_legacy_agent_miner_control, "miner-1", {"role": "Admin"}),
+            (main.process_due_dashboard_scheduled_bulk_actions, {"role": "Admin"}),
+            (main.create_dashboard_scheduled_bulk_action, {"role": "Admin"}),
+            (main.whatsminer_read_test, {"role": "Admin"}),
+        )
+
+        with (
+            patch.object(main, "get_db_connection") as sync_database,
+            patch.object(main, "minerops_command_db_connect") as async_database,
+            patch.object(main, "insert_miner_command_event") as audit_write,
+            patch.object(main, "minerops_send_whatsminer_reboot") as reboot_write,
+            patch.object(main, "minerops_power_command") as power_write,
+            patch.object(main, "minerops_api3_change_password") as password_write,
+        ):
+            for route in routes:
+                with self.subTest(route=route[0].__name__):
+                    self.assert_retired(route[0], *route[1:])
+
+        for forbidden_call in (
+            sync_database,
+            async_database,
+            audit_write,
+            reboot_write,
+            power_write,
+            password_write,
+        ):
+            self.assertFalse(forbidden_call.called)
+
+    def test_m30_requires_control_permission_before_inert_response(self):
+        self.assertEqual(
+            required_api_permission(
+                "POST", "/api/dashboard/m30-command-test-lab/run"
+            ),
+            "miners.control",
+        )
+        dependency = require_permission("miners.control")
+        with self.assertRaises(HTTPException) as denied:
+            dependency({"role": "Read Only", "permissions": []})
+        self.assertEqual(denied.exception.status_code, 403)
+
+        route_source = inspect.getsource(main.run_dashboard_m30_command_test_lab)
+        self.assertIn('Depends(require_permission("miners.control"))', route_source)
+        self.assertIn("status_code=410", route_source)
+        self.assertNotIn("minerops_agent_command_queue", route_source)
+        self.assertNotIn("plexus-windows-agent", route_source)
+        self.assertNotIn("dry_run", route_source)
+
+        app = FastAPI()
+        app.add_api_route(
+            "/api/dashboard/m30-command-test-lab/run",
+            main.run_dashboard_m30_command_test_lab,
+            methods=["POST"],
+        )
+        with TestClient(app) as client:
+            app.dependency_overrides[get_current_user] = lambda: {
+                "role": "Read Only",
+                "permissions": [],
+            }
+            self.assertEqual(
+                client.post("/api/dashboard/m30-command-test-lab/run").status_code,
+                403,
+            )
+            app.dependency_overrides[get_current_user] = lambda: {
+                "role": "Custom",
+                "permissions": ["miners.control"],
+            }
+            self.assertEqual(
+                client.post("/api/dashboard/m30-command-test-lab/run").status_code,
+                410,
+            )
+
+    def test_registered_retired_routes_have_no_direct_write_implementation(self):
+        retired_paths = {
+            "/api/dashboard/m30-command-test-lab/run",
+            "/api/dashboard/scheduled-bulk-actions/process-due",
+            "/api/dashboard/scheduled-bulk-actions",
+            "/api/tools/whatsminer-read-test",
+            "/api/miners/{miner_id}/commands/reboot",
+            "/api/miners/{miner_id}/commands/power-off",
+            "/api/miners/{miner_id}/commands/power-on",
+            "/api/miners/{miner_id}/commands/suspend",
+            "/api/miners/{miner_id}/commands/resume",
+            "/api/miners/{miner_id}/api3/change-password",
+            "/api/miners/{miner_id}/agent-control",
+        }
+        registered = [
+            (route.path, route.endpoint)
+            for route in main.app.routes
+            if getattr(route, "path", None) in retired_paths
+            and "POST" in getattr(route, "methods", set())
+        ]
+        self.assertEqual({path for path, _ in registered}, retired_paths)
+        self.assertEqual(len(registered), len(retired_paths))
+
+        forbidden = (
+            "socket",
+            "minerops_send_",
+            "minerops_power_command",
+            "minerops_api3_change_password",
+            "insert_miner_command_event",
+            "minerops_agent_command_queue",
+            "minerops_command_db_connect",
+            "get_db_connection",
+            "request.json",
+        )
+        for path, endpoint in registered:
+            source = inspect.getsource(endpoint)
+            with self.subTest(path=path):
+                self.assertIn("status_code=410", source)
+                for unsafe_fragment in forbidden:
+                    self.assertNotIn(unsafe_fragment, source)
+
+    def test_legacy_scheduler_is_noop_and_not_registered_before_claim(self):
+        with patch.object(main, "get_db_connection") as database:
+            self.assertEqual(main.minerops_process_due_scheduled_bulk_actions(), [])
+        self.assertFalse(database.called)
+        self.assertFalse(hasattr(main.app.state, "agent_command_before_claim"))
+        source = inspect.getsource(main.minerops_process_due_scheduled_bulk_actions)
+        self.assertNotIn("minerops_agent_command_queue", source)
+        self.assertNotIn("get_db_connection", source)
+
+    def test_every_retired_http_route_denies_unprivileged_users_and_never_writes(self):
+        paths = {
+            "/api/dashboard/m30-command-test-lab/run",
+            "/api/dashboard/scheduled-bulk-actions",
+            "/api/dashboard/scheduled-bulk-actions/process-due",
+            "/api/tools/whatsminer-read-test",
+            "/api/miners/{miner_id}/commands/reboot",
+            "/api/miners/{miner_id}/commands/power-off",
+            "/api/miners/{miner_id}/commands/power-on",
+            "/api/miners/{miner_id}/commands/suspend",
+            "/api/miners/{miner_id}/commands/resume",
+            "/api/miners/{miner_id}/api3/change-password",
+            "/api/miners/{miner_id}/agent-control",
+        }
+        app = FastAPI()
+        for route in main.app.routes:
+            if getattr(route, "path", None) in paths and "POST" in route.methods:
+                app.add_api_route(route.path, route.endpoint, methods=["POST"])
+        with TestClient(app) as client, patch.object(main.socket, "create_connection") as transport, patch.object(main, "get_db_connection") as database:
+            for permissions, expected in (([], 403), (["miners.control", "credentials.manage"], 410)):
+                app.dependency_overrides[get_current_user] = lambda: {
+                    "role": "Custom", "permissions": permissions,
+                }
+                for path in paths:
+                    concrete = path.replace("{miner_id}", "miner-1")
+                    self.assertEqual(client.post(concrete, json={"dry_run": False}).status_code, expected, concrete)
+            transport.assert_not_called()
+            database.assert_not_called()
+
+    def test_diagnostic_write_disguises_overrides_and_api3_are_retired(self):
+        app = FastAPI()
+        app.add_api_route("/api/tools/whatsminer-read-test", main.whatsminer_read_test, methods=["POST"])
+        app.dependency_overrides[get_current_user] = lambda: {"role": "Custom", "permissions": ["miners.control"]}
+        attacks = (
+            {"command": "reboot"}, {"command": "read_then_reboot"},
+            {"command": "summary", "params": {"command": "reboot"}},
+            {"command": "summary", "params": {"cmd": "set.miner.service", "param": "stop"}},
+            {"port": 4433, "command": "set.miner.power_mode", "params": {"param": "high"}},
+            {"command": "summary\n{\"command\":\"reboot\"}"},
+            {"command": "summary"},
+        )
+        with TestClient(app) as client, patch.object(main.socket, "create_connection") as transport, patch.object(main, "send_whatsminer_read_command") as sender:
+            for attack in attacks:
+                self.assertEqual(client.post("/api/tools/whatsminer-read-test", json={"ip_address": "192.0.2.1", **attack}).status_code, 410)
+            transport.assert_not_called()
+            sender.assert_not_called()
+
+    def test_internal_discovery_helper_cannot_become_a_generic_command_gateway(self):
+        with patch.object(main.socket, "create_connection") as transport:
+            for command, port, params in (
+                ("reboot", 4028, None), ("read_reboot", 4028, None),
+                ("summary", 4433, None), ("set.miner.service", 4433, {"param": "stop"}),
+                ("summary", 4028, {"command": "reboot"}),
+                ("summary", 4028, {"cmd": "set.miner.service"}),
+                ("get_miner_info", 4028, {"info": "arbitrary"}),
+            ):
+                self.assert_retired(main.send_whatsminer_read_command, "192.0.2.1", port, command, 500, params)
+            transport.assert_not_called()
+
+    def test_old_agent_aliases_are_inert_even_without_authentication(self):
+        app = FastAPI()
+        app.add_api_route("/api/agent/commands/pending-disabled-old", main.minerops_agent_commands_pending_retired, methods=["GET"])
+        app.add_api_route("/api/agent/commands/result-disabled-old", main.minerops_agent_commands_result_retired, methods=["POST"])
+        with TestClient(app) as client, patch.object(main, "minerops_command_db_connect") as database:
+            self.assertEqual(client.get("/api/agent/commands/pending-disabled-old").status_code, 410)
+            self.assertEqual(client.post("/api/agent/commands/result-disabled-old", json={"status": "completed"}).status_code, 410)
+            database.assert_not_called()
+
+    def test_canonical_control_still_derives_agent_from_miner_zone(self):
+        repository_source = (
+            APP_ROOT / "miner_control" / "repository.py"
+        ).read_text()
+        self.assertIn("miners.agent_zone_id", repository_source)
+        self.assertIn(
+            "agents.agent_zone_id = miners.agent_zone_id",
+            repository_source,
+        )
+        self.assertNotIn("agents.site_id = miners.site_id", repository_source)
+
+
+if __name__ == "__main__":
+    unittest.main()

backend/tests/test_miner_control.py

@@ -71,64 +71,47 @@ class FakeRepository:
 
 
 class MinerControlServiceTests(unittest.TestCase):
-    def test_supported_projection_is_factual_and_bounded(self):
+    def test_disabled_projection_preserves_facts_and_bounded_history(self):
         repository = FakeRepository()
         result = MinerControlService(repository).get_controls(MINER_ID)
-        self.assertEqual(result.service_control.state, "supported")
+        self.assertEqual(result.service_control.state, "unavailable")
+        self.assertEqual(result.service_control.reason, "physical_control_disabled_stage_a")
         self.assertEqual(result.performance_mode, "normal")
-        self.assertEqual(result.available_service_action, "suspend")
+        self.assertIsNone(result.available_service_action)
         self.assertEqual(repository.limits, [5])
 
-    def test_capability_missing_facts_is_not_optimistic(self):
-        cases = (
-            ({"agent_zone_id": None}, "unavailable", "agent_zone_unassigned"),
-            ({"agent_zone_exists": False}, "unavailable", "agent_zone_missing"),
-            ({"agent_zone_enabled": False}, "unavailable", "agent_zone_disabled"),
-            ({"agent_id": None}, "unavailable", "agent_unavailable"),
-            ({"current_ip": None}, "unavailable", "current_ip_missing"),
-            ({"api3_supported": False}, "unavailable", "api3_unsupported"),
-            ({"api3_supported": None}, "unknown", "api3_support_unknown"),
-            ({"last_payload": {}}, "unknown", "agent_capability_unknown"),
-            (
-                {"last_payload": {"control_available": False,
-                                  "control_unavailable_reason": "credentials_missing"}},
-                "unavailable", "credentials_missing",
-            ),
-        )
-        for changes, state, reason in cases:
-            with self.subTest(reason=reason):
-                result = MinerControlService(FakeRepository(context(**changes))).get_controls(MINER_ID)
-                self.assertEqual(result.service_control.state, state)
-                self.assertEqual(result.service_control.reason, reason)
+    def test_capability_cannot_be_enabled_by_agent_claims(self):
+        for facts in ({}, {"control_available": True}, {"control_available": False}):
+            repository = FakeRepository(context(last_payload=facts))
+            result = MinerControlService(repository).get_controls(MINER_ID)
+            self.assertEqual(result.service_control.state, "unavailable")
+            self.assertEqual(result.service_control.reason, "physical_control_disabled_stage_a")
 
-    def test_resume_requires_known_stopped_state(self):
-        service = MinerControlService(FakeRepository(context(runtime_state="offline")))
-        with self.assertRaisesRegex(MinerControlUnavailable, "factually_stopped"):
-            service.queue_control(
-                MINER_ID, "resume", requested_value=None,
-                actor_user_id=USER_ID, actor_username="operator",
-            )
+    def test_resume_is_disabled_even_for_known_stopped_state(self):
+        service = MinerControlService(FakeRepository(context(runtime_state="stopped")))
+        with self.assertRaisesRegex(MinerControlUnavailable, "physical_control_disabled_stage_a"):
+            service.queue_control(MINER_ID, "resume", requested_value=None,
+                                  actor_user_id=USER_ID, actor_username="operator")
 
-    def test_queue_contains_routing_but_audit_and_response_do_not(self):
+    def test_all_current_actions_are_rejected_before_routing_or_queue_write(self):
+        for action in ("suspend", "resume", "performance_mode"):
+            repository = FakeRepository()
+            with self.assertRaisesRegex(MinerControlUnavailable, "physical_control_disabled_stage_a"):
+                MinerControlService(repository).queue_control(
+                    MINER_ID, action, requested_value="high",
+                    actor_user_id=USER_ID, actor_username="operator")
+            self.assertEqual(repository.enqueue_calls, [])
+            self.assertFalse(hasattr(repository, "requested_miner"))
+
+    def test_rejected_request_leaves_context_unchanged(self):
         repository = FakeRepository()
-        result = MinerControlService(repository).queue_control(
-            MINER_ID, "performance_mode", requested_value="high",
-            actor_user_id=USER_ID, actor_username="operator",
-        )
-        call = repository.enqueue_calls[0]
-        self.assertEqual(call["request_payload"]["ip_address"], "192.0.2.10")
-        self.assertNotIn("ip_address", call["audit_payload"])
-        self.assertNotIn("agent_id", call["audit_payload"])
-        self.assertNotIn("password", str(call).lower())
-        self.assertNotIn("192.0.2.10", result.model_dump_json())
-
-    def test_no_projection_changes_when_work_is_queued(self):
-        repository = FakeRepository()
-        MinerControlService(repository).queue_control(
-            MINER_ID, "suspend", requested_value=None,
-            actor_user_id=USER_ID, actor_username="operator",
-        )
-        self.assertFalse(any("update" in key for key in repository.enqueue_calls[0]))
+        before = repository.miner_context.copy()
+        with self.assertRaises(MinerControlUnavailable):
+            MinerControlService(repository).queue_control(
+                MINER_ID, "suspend", requested_value=None,
+                actor_user_id=USER_ID, actor_username="operator")
+        self.assertEqual(repository.miner_context, before)
+        self.assertEqual(repository.enqueue_calls, [])
 
 
 class MinerControlRouterTests(unittest.TestCase):
@@ -159,18 +142,16 @@ class MinerControlRouterTests(unittest.TestCase):
         self.assertEqual(response.status_code, 200)
         self.assertEqual(response.json()["data"]["miner_id"], str(MINER_ID))
 
-    def test_posts_require_control_and_return_202_bounded_envelope(self):
-        path = f"/api/miners/{MINER_ID}/controls/suspend"
-        self.authenticate(("miners.view",))
-        self.assertEqual(self.client.post(path, json={}).status_code, 403)
-        self.authenticate(("miners.control",))
-        response = self.client.post(path, json={})
-        self.assertEqual(response.status_code, 202)
-        self.assertEqual(set(response.json()["data"]), {
-            "operation_id", "miner_id", "action", "requested_value", "status", "created_at"
-        })
-        self.assertNotIn("ip", response.text.lower())
-        self.assertNotIn("agent", response.text.lower())
+    def test_posts_require_control_and_then_fail_closed(self):
+        for action, body in (("suspend", {}), ("resume", {}), ("performance-mode", {"mode": "high"})):
+            path = f"/api/miners/{MINER_ID}/controls/{action}"
+            self.authenticate(("miners.view",))
+            self.assertEqual(self.client.post(path, json=body).status_code, 403)
+            self.authenticate(("miners.control",))
+            response = self.client.post(path, json=body)
+            self.assertEqual(response.status_code, 409)
+            self.assertIn("physical_control_disabled_stage_a", response.text)
+        self.assertEqual(self.repository.enqueue_calls, [])
 
     def test_strict_bodies_reject_routing_commands_and_credentials(self):
         self.authenticate(("miners.control",))
@@ -195,7 +176,7 @@ class MinerControlRouterTests(unittest.TestCase):
         )
         self.assertEqual(
             self.client.post(path, json={}, headers={"Origin": "http://testserver"}).status_code,
-            202,
+            409,
         )
 
     def test_identity_permission_mapping_is_specific(self):

backend/tests/test_miner_control_postgres.py

@@ -137,7 +137,7 @@ class MinerControlPostgresTests(unittest.TestCase):
                     """
                     INSERT INTO minerops_agent_command_queue (
                         id, agent_id, command_type, command_status, request_payload, created_at
-                    ) VALUES (%s, %s, 'miner_control', %s, %s, COALESCE(%s, now()))
+                    ) VALUES (%s, %s, 'poll_known_miners_now', %s, %s, COALESCE(%s, now()))
                     """,
                     (command_id, AGENT_ID, status,
                      Jsonb({"schema_version": 1, "miner_id": str(miner_id),
@@ -199,13 +199,22 @@ class MinerControlPostgresTests(unittest.TestCase):
             service.submit_result(AGENT_ID, pending)
 
     def test_atomic_queue_audit_duplicate_live_and_completed_history(self):
-        service = MinerControlService(MinerControlRepository(connect))
-        first = service.queue_control(
+        repository = MinerControlRepository(connect)
+        def queue_control(miner_id, action, *, requested_value, actor_user_id, actor_username):
+            from types import SimpleNamespace
+            from uuid import uuid4
+            return SimpleNamespace(**repository.enqueue_with_audit(
+                operation_id=str(uuid4()), miner_id=miner_id, agent_id=AGENT_ID,
+                actor_user_id=actor_user_id, actor_username=actor_username,
+                request_payload={"schema_version": 1, "miner_id": str(miner_id),
+                                 "action": action, "requested_value": requested_value},
+                audit_payload={"miner_id": str(miner_id), "action": action}))
+        first = queue_control(
             MINER_ID, "suspend", requested_value=None,
             actor_user_id=USER_ID, actor_username="control-test-user",
         )
         with self.assertRaises(MinerControlConflict):
-            service.queue_control(
+            queue_control(
                 MINER_ID, "suspend", requested_value=None,
                 actor_user_id=USER_ID, actor_username="control-test-user",
             )
@@ -223,17 +232,26 @@ class MinerControlPostgresTests(unittest.TestCase):
             connection.commit()
         finally:
             connection.close()
-        second = service.queue_control(
+        second = queue_control(
             MINER_ID, "suspend", requested_value=None,
             actor_user_id=USER_ID, actor_username="control-test-user",
         )
         self.assertNotEqual(first.operation_id, second.operation_id)
 
     def test_audit_failure_rolls_back_queue_row(self):
-        service = MinerControlService(MinerControlRepository(connect))
+        repository = MinerControlRepository(connect)
+        def queue_control(miner_id, action, *, requested_value, actor_user_id, actor_username):
+            from types import SimpleNamespace
+            from uuid import uuid4
+            return SimpleNamespace(**repository.enqueue_with_audit(
+                operation_id=str(uuid4()), miner_id=miner_id, agent_id=AGENT_ID,
+                actor_user_id=actor_user_id, actor_username=actor_username,
+                request_payload={"schema_version": 1, "miner_id": str(miner_id),
+                                 "action": action, "requested_value": requested_value},
+                audit_payload={"miner_id": str(miner_id), "action": action}))
         operation_count_before = self._operation_count()
         with self.assertRaises(Exception):
-            service.queue_control(
+            queue_control(
                 MINER_ID, "suspend", requested_value=None,
                 actor_user_id=UUID("00000000-0000-4000-8000-000000009999"),
                 actor_username="missing-user",

backend/tests/test_stage_a_containment.py

@@ -0,0 +1,89 @@
+"""Defense-in-depth unit tests; no physical miner or external service is used."""
+
+import asyncio
+import inspect
+import unittest
+from datetime import datetime, timezone
+from unittest.mock import Mock, patch
+
+from fastapi import HTTPException
+
+from app import main
+from app.agent_commands.exceptions import AgentCommandConflict
+from app.agent_commands.models import AgentCommandResultRequest
+from app.agent_commands.policy import rejection_reason
+from app.agent_commands.service import AgentCommandService
+
+
+class StageAContainmentTests(unittest.TestCase):
+    def test_service_dispatch_rejects_physical_and_unknown_rows_even_if_repository_regresses(self):
+        repository = Mock()
+        repository.claim_pending.return_value = [
+            {"command_id": str(index), "command_type": kind, "request_payload": payload,
+             "created_at": datetime.now(timezone.utc)}
+            for index, (kind, payload) in enumerate((
+                ("miner_control", {"schema_version": 1, "action": "suspend"}),
+                ("miner_control", {"schema_version": 999}),
+                ("miner_control_v2", {}), ("raw_api3", {}),
+                ("poll_known_miners_now", {}),
+            ))
+        ]
+        result = AgentCommandService(repository).claim("simulated-agent")
+        self.assertEqual([row.command_type for row in result], ["poll_known_miners_now"])
+
+    def test_service_result_rejection_does_not_depend_on_payload_validation(self):
+        for kind in ("miner_control", "raw_api3", "miner_control_v2"):
+            repository = Mock()
+            repository.get_command.return_value = {"id": "old-command", "agent_id": "simulated-agent",
+                "command_type": kind, "command_status": "claimed", "request_payload": {}}
+            with self.assertRaises(AgentCommandConflict):
+                AgentCommandService(repository).submit_result("simulated-agent",
+                    AgentCommandResultRequest(command_id="old-command", status="completed",
+                                              result_payload={"schema_version": 999, "command_accepted": True}))
+            args = repository.finish_command.call_args.args
+            self.assertEqual(args[2:4], ("cancelled", {}))
+            self.assertEqual(args[4], rejection_reason(kind))
+
+    def test_dormant_legacy_helpers_cannot_transmit_or_queue_if_called_internally(self):
+        names = (
+            "dry_run_reboot_miner_command", "change_miner_api3_password",
+            "minerops_queue_agent_miner_control", "minerops_power_command",
+            "minerops_send_whatsminer_reboot_sync", "minerops_send_whatsminer_reboot",
+            "minerops_send_whatsminer_command_sync", "minerops_api3_send_payload_sync",
+            "minerops_api3_service_sync", "minerops_api3_service",
+            "minerops_send_control_command_item", "minerops_send_whatsminer_command_sequence",
+            "minerops_send_whatsminer_command", "minerops_api3_service_sync_with_credentials",
+            "minerops_api3_service_with_credentials", "minerops_api3_change_password_sync",
+            "minerops_api3_change_password",
+        )
+        with patch.object(main.socket, "create_connection") as transport, patch.object(main, "get_db_connection") as database, patch.object(main, "minerops_command_db_connect") as async_database:
+            for name in names:
+                function = getattr(main, name)
+                required = ["simulated" for parameter in inspect.signature(function).parameters.values()
+                            if parameter.default is inspect.Parameter.empty]
+                with self.subTest(name=name), self.assertRaises(HTTPException) as rejected:
+                    result = function(*required)
+                    if inspect.isawaitable(result):
+                        asyncio.run(result)
+                self.assertEqual(rejected.exception.status_code, 410)
+            transport.assert_not_called()
+            database.assert_not_called()
+            async_database.assert_not_called()
+
+    def test_fixed_internal_discovery_queries_still_use_only_allowlisted_selection(self):
+        parameters = {"get_miner_info": {"info": "ip,proto,netmask,gateway,dns,hostname,mac,ledstat,minersn,powersn"},
+                      "get_version": {}, "summary": {}, "edevs": {}, "pools": {}}
+        for command, params in parameters.items():
+            with patch.object(main.socket, "create_connection") as transport:
+                socket = transport.return_value.__enter__.return_value
+                socket.recv.side_effect = [b'{"STATUS":[]}', b'']
+                result = main.send_whatsminer_read_command("192.0.2.1", 4028, command, 500, params)
+                self.assertTrue(result["success"])
+                self.assertEqual(socket.sendall.call_count, 1)
+                import json
+                transmitted = json.loads(socket.sendall.call_args.args[0])
+                self.assertEqual(transmitted, {"command": command, **params})
+
+
+if __name__ == "__main__":
+    unittest.main()

backend/tests/test_stage_a_containment_postgres.py

@@ -0,0 +1,191 @@
+"""Stage A queue containment; requires an explicitly disposable PostgreSQL DB."""
+
+import os
+import unittest
+from concurrent.futures import ThreadPoolExecutor
+from unittest.mock import patch
+
+import psycopg
+from psycopg.rows import dict_row
+from psycopg.types.json import Jsonb
+
+from app.agent_commands.exceptions import AgentCommandConflict
+from app.agent_commands.models import AgentCommandResultRequest
+from app.agent_commands.repository import AgentCommandRepository
+from app.agent_commands.service import AgentCommandService
+from app.config import settings
+
+
+AGENT = "stage-a-simulated-agent"
+OTHER_AGENT = "stage-a-other-agent"
+
+
+def connect():
+    return psycopg.connect(host=settings.postgres_host, port=settings.postgres_port,
+                          dbname=settings.postgres_db, user=settings.postgres_user,
+                          password=settings.postgres_password, connect_timeout=5)
+
+
+class StageAQueueContainmentTests(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        if os.environ.get("MINEROPS_STAGE_A_DISPOSABLE") != "1" or not settings.postgres_db.startswith("minerops_stage_a_"):
+            raise RuntimeError("Refusing queue tests outside an explicitly disposable Stage A database")
+
+    def setUp(self):
+        self.transport_patch = patch("socket.create_connection", side_effect=AssertionError("No miner transport allowed"))
+        self.transport = self.transport_patch.start()
+        self.addCleanup(self.transport_patch.stop)
+        self.addCleanup(self.transport.assert_not_called)
+        with connect() as connection, connection.cursor() as cur:
+            cur.execute("DELETE FROM audit_logs WHERE entity_type = 'agent_command' AND old_value_json ->> 'agent_id' IN (%s, %s)", (AGENT, OTHER_AGENT))
+            cur.execute("DELETE FROM minerops_agent_command_queue WHERE agent_id IN (%s, %s)", (AGENT, OTHER_AGENT))
+        self.repository = AgentCommandRepository(connect)
+        self.service = AgentCommandService(self.repository)
+
+    @staticmethod
+    def seed(command_id, status="pending", command_type="miner_control", payload=None, agent=AGENT):
+        if payload is None:
+            payload = {"action": "legacy_4028", "ip_address": "192.0.2.1", "legacy_4028_command": "reboot", "dry_run": False}
+        with connect() as connection, connection.cursor() as cur:
+            cur.execute("""INSERT INTO minerops_agent_command_queue
+                (id, agent_id, command_type, command_status, request_payload,
+                 result_payload, error_message, claimed_at)
+                VALUES (%s, %s, %s, %s, %s, %s, 'prior evidence',
+                        CASE WHEN %s = 'claimed' THEN now() ELSE NULL END)""",
+                (command_id, agent, command_type, status, Jsonb(payload),
+                 Jsonb({"historical_evidence": True}), status))
+
+    @staticmethod
+    def row(command_id):
+        with connect() as connection, connection.cursor(row_factory=dict_row) as cur:
+            cur.execute("SELECT * FROM minerops_agent_command_queue WHERE id = %s", (command_id,))
+            return dict(cur.fetchone())
+
+    @staticmethod
+    def audits(command_id):
+        with connect() as connection, connection.cursor(row_factory=dict_row) as cur:
+            cur.execute("SELECT * FROM audit_logs WHERE action = 'agent.command.rejected' AND old_value_json ->> 'command_id' = %s", (command_id,))
+            return [dict(row) for row in cur.fetchall()]
+
+    def assert_cancelled_preserving_history(self, command_id, before, outcome):
+        after = self.row(command_id)
+        self.assertEqual(after["command_status"], "cancelled")
+        for field in ("id", "agent_id", "command_type", "request_payload", "result_payload", "created_at", "claimed_at"):
+            self.assertEqual(after[field], before[field], field)
+        self.assertTrue(after["error_message"].startswith("prior evidence\n"))
+        self.assertIsNotNone(after["completed_at"])
+        audits = self.audits(command_id)
+        self.assertEqual(len(audits), 1)
+        self.assertEqual(audits[0]["old_value_json"]["status"], before["command_status"])
+        self.assertEqual(audits[0]["new_value_json"]["execution_outcome"], outcome)
+        self.assertNotIn("192.0.2.1", str(audits))
+
+    def test_seeded_legacy_pending_never_dispatched(self):
+        self.seed("stage-a-pending")
+        before = self.row("stage-a-pending")
+        self.assertEqual(self.service.claim(AGENT), [])
+        self.assert_cancelled_preserving_history("stage-a-pending", before, "unknown")
+
+    def test_seeded_legacy_claimed_cancelled_with_unknown_prior_execution(self):
+        self.seed("stage-a-claimed", status="claimed")
+        before = self.row("stage-a-claimed")
+        self.assertEqual(self.service.claim(AGENT), [])
+        self.assert_cancelled_preserving_history("stage-a-claimed", before, "unknown")
+
+    def test_all_physical_schemas_and_unknown_types_are_rejected(self):
+        payloads = ({}, {"schema_version": 1, "action": "suspend"},
+                    {"schema_version": 2, "action": "resume"},
+                    {"schema_version": "1", "action": "performance_mode"},
+                    {"schema_version": 999, "action": "reboot"})
+        for index, payload in enumerate(payloads):
+            self.seed(f"stage-a-schema-{index}", payload=payload)
+        self.seed("stage-a-unknown", command_type="raw_api3")
+        self.assertEqual(self.service.claim(AGENT), [])
+        for index in range(len(payloads)):
+            self.assertIn("physical_control_disabled_stage_a", self.row(f"stage-a-schema-{index}")["error_message"])
+        self.assertIn("unsupported_command_schema_stage_a", self.row("stage-a-unknown")["error_message"])
+
+    def test_late_results_and_replay_cannot_promote_old_claimed_commands(self):
+        self.seed("stage-a-late", status="claimed")
+        before = self.row("stage-a-late")
+        for status in ("completed", "failed", "completed"):
+            request = AgentCommandResultRequest(command_id="stage-a-late", status=status,
+                                                result_payload={"command_accepted": True})
+            with self.assertRaisesRegex(AgentCommandConflict, "physical_control_disabled_stage_a"):
+                self.service.submit_result(AGENT, request)
+        self.assert_cancelled_preserving_history("stage-a-late", before, "unknown")
+
+    def test_repository_cannot_be_bypassed_by_direct_finish_call(self):
+        self.seed("stage-a-direct", status="pending")
+        before = self.row("stage-a-direct")
+        with self.assertRaises(AgentCommandConflict):
+            self.repository.finish_command("stage-a-direct", AGENT, "completed", {"accepted": True}, None)
+        self.assert_cancelled_preserving_history("stage-a-direct", before, "unknown")
+
+    def test_wrong_agent_cannot_cancel_or_promote_another_agents_entry(self):
+        self.seed("stage-a-foreign", status="claimed", agent=OTHER_AGENT)
+        before = self.row("stage-a-foreign")
+        self.assertEqual(self.service.claim(AGENT), [])
+        with self.assertRaises(AgentCommandConflict):
+            self.service.submit_result(AGENT, AgentCommandResultRequest(command_id="stage-a-foreign", status="completed"))
+        with self.assertRaises(AgentCommandConflict):
+            self.repository.finish_command("stage-a-foreign", AGENT, "completed", {}, None)
+        self.assertEqual(self.row("stage-a-foreign"), before)
+        self.assertEqual(self.audits("stage-a-foreign"), [])
+
+    def test_terminal_history_is_immutable_and_not_reaudited(self):
+        for status in ("completed", "failed", "cancelled"):
+            command_id = f"stage-a-terminal-{status}"
+            self.seed(command_id, status=status)
+            before = self.row(command_id)
+            self.assertEqual(self.service.claim(AGENT), [])
+            with self.assertRaises(AgentCommandConflict):
+                self.repository.finish_command(command_id, AGENT, "completed", {}, None)
+            self.assertEqual(self.row(command_id), before)
+            self.assertEqual(self.audits(command_id), [])
+
+    def test_read_only_commands_still_claim_and_complete(self):
+        for kind in ("poll_known_miners_now", "discover_new_miners", "enrich_match_discovery"):
+            self.seed(f"stage-a-read-{kind}", command_type=kind, payload={"ip_address": "192.0.2.1"})
+        commands = self.service.claim(AGENT)
+        self.assertEqual(len(commands), 3)
+        for command in commands:
+            self.service.submit_result(AGENT, AgentCommandResultRequest(command_id=command.command_id, status="completed", result_payload={"simulated": True}))
+            self.assertEqual(self.row(command.command_id)["command_status"], "completed")
+
+    def test_concurrent_pollers_cancel_once_and_never_deliver_physical_commands(self):
+        self.seed("stage-a-concurrent")
+        before = self.row("stage-a-concurrent")
+        with ThreadPoolExecutor(max_workers=2) as pool:
+            results = list(pool.map(lambda _: AgentCommandRepository(connect).claim_pending(AGENT), range(2)))
+        self.assertEqual(results, [[], []])
+        self.assert_cancelled_preserving_history("stage-a-concurrent", before, "unknown")
+
+    def test_locked_legacy_entry_cannot_leak_into_claim_query(self):
+        self.seed("stage-a-locked")
+        locker = connect()
+        try:
+            with locker.cursor() as cur:
+                cur.execute("SELECT id FROM minerops_agent_command_queue WHERE id = 'stage-a-locked' FOR UPDATE")
+                self.assertEqual(self.service.claim(AGENT), [])
+        finally:
+            locker.rollback()
+            locker.close()
+        self.assertEqual(self.service.claim(AGENT), [])
+        self.assertEqual(self.row("stage-a-locked")["command_status"], "cancelled")
+
+    def test_audit_failure_rolls_back_cancellation_and_prevents_dispatch(self):
+        self.seed("stage-a-audit-failure")
+        before = self.row("stage-a-audit-failure")
+        def fail_audit(cur, row, **kwargs):
+            cur.execute("SELECT 1 / 0")
+        with patch.object(AgentCommandRepository, "_audit_rejection", side_effect=fail_audit):
+            with self.assertRaises(psycopg.errors.DivisionByZero):
+                self.service.claim(AGENT)
+        self.assertEqual(self.row("stage-a-audit-failure"), before)
+        self.assertEqual(self.audits("stage-a-audit-failure"), [])
+
+
+if __name__ == "__main__":
+    unittest.main()

backend/tests/test_stage_a_delivery_evidence.py

@@ -0,0 +1,21 @@
+"""Conservative delivery outcome decisions, independent of SQL row shape."""
+import unittest
+from app.agent_commands.policy import rejection_execution_outcome
+
+
+class DeliveryOutcomeTests(unittest.TestCase):
+    def test_only_explicit_clean_pending_evidence_allows_not_dispatched(self):
+        self.assertEqual(rejection_execution_outcome({"previous_status": "pending", "no_delivery_evidence": True}), "not_dispatched")
+
+    def test_missing_ambiguous_or_reported_delivery_is_unknown(self):
+        for row in ({}, {"previous_status": "pending"},
+                    {"previous_status": "pending", "no_delivery_evidence": False},
+                    {"previous_status": "pending", "no_delivery_evidence": None},
+                    {"previous_status": "pending", "no_delivery_evidence": 1},
+                    {"previous_status": "pending", "no_delivery_evidence": "true"},
+                    {"previous_status": "claimed", "no_delivery_evidence": True},
+                    {"previous_status": "cancelled", "no_delivery_evidence": True},
+                    {"previous_status": "pending", "no_delivery_evidence": True, "delivery_reported": True},
+                    {"previous_status": "pending", "no_delivery_evidence": True, "delivery_reported": None}):
+            with self.subTest(row=row):
+                self.assertEqual(rejection_execution_outcome(row), "unknown")

backend/tests/test_stage_a_delivery_evidence_postgres.py

@@ -0,0 +1,230 @@
+"""Delivery evidence across every Stage A retirement entry point; no miner IO."""
+import os
+import unittest
+import time
+from concurrent.futures import ThreadPoolExecutor
+from unittest.mock import patch
+from psycopg.rows import dict_row
+from psycopg.types.json import Jsonb
+from app.agent_commands.exceptions import AgentCommandConflict
+from app.agent_commands.repository import AgentCommandRepository
+from app.config import settings
+from tests.test_stage_a_containment_postgres import connect
+
+AGENT = "stage-a-final-evidence-agent"
+ACTOR = "00000000-0000-4000-8000-000000003302"
+SCHEDULE = "stage-a-final-evidence-schedule"
+PATHS = ("poll", "result", "cutover", "cancel", "delete")
+# Verbatim CLAIM_COMMANDS_SQL from committed 2d89e80. Test-only legacy delivery;
+# never use this query in the Stage A production repository.
+BASELINE_CLAIM_SQL = """
+WITH candidates AS (
+    SELECT id
+    FROM minerops_agent_command_queue
+    WHERE agent_id = %s
+      AND command_status = 'pending'
+    ORDER BY created_at, id
+    FOR UPDATE SKIP LOCKED
+    LIMIT %s
+)
+UPDATE minerops_agent_command_queue AS queue
+SET command_status = 'claimed', claimed_at = now()
+FROM candidates
+WHERE queue.id = candidates.id
+  AND queue.command_status = 'pending'
+RETURNING queue.id AS command_id, queue.command_type,
+          queue.request_payload, queue.created_at
+"""
+
+
+class DeliveryEvidencePostgresTests(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        if os.environ.get("MINEROPS_STAGE_A_DISPOSABLE") != "1" or not settings.postgres_db.startswith("minerops_stage_a_"):
+            raise RuntimeError("Delivery evidence tests require a disposable Stage A database")
+
+    def setUp(self):
+        self.repo = AgentCommandRepository(connect)
+        self.clean()
+        self.addCleanup(self.clean)
+        with connect() as conn, conn.cursor() as cur:
+            cur.execute("""INSERT INTO users (id,username,display_name,password_hash,role_id)
+                           SELECT %s,'stage-a-final-evidence','Delivery Evidence','unused',id
+                           FROM roles WHERE name='Admin' ON CONFLICT (id) DO NOTHING""", (ACTOR,))
+        self.no_transport = patch("socket.create_connection", side_effect=AssertionError("No miner IO"))
+        mocked = self.no_transport.start()
+        self.addCleanup(mocked.assert_not_called)
+        self.addCleanup(self.no_transport.stop)
+
+    def clean(self):
+        with connect() as conn, conn.cursor() as cur:
+            cur.execute("DELETE FROM audit_logs WHERE actor_user_id=%s OR old_value_json->>'agent_id'=%s", (ACTOR, AGENT))
+            cur.execute("DELETE FROM minerops_agent_command_results WHERE agent_id=%s", (AGENT,))
+            cur.execute("DELETE FROM minerops_agent_command_queue WHERE agent_id=%s", (AGENT,))
+            cur.execute("DELETE FROM minerops_scheduled_bulk_actions WHERE id=%s", (SCHEDULE,))
+
+    def seed(self, command_id="stage-a-final-evidence-command", kind="miner_control"):
+        with connect() as conn, conn.cursor() as cur:
+            cur.execute("""INSERT INTO minerops_scheduled_bulk_actions
+                           (id,agent_id,target_type,status,scheduled_for) VALUES (%s,%s,'all','running',now())
+                           ON CONFLICT (id) DO UPDATE SET status='running'""", (SCHEDULE,AGENT))
+            cur.execute("""INSERT INTO minerops_agent_command_queue (id,agent_id,command_type,request_payload)
+                           VALUES (%s,%s,%s,%s)""",
+                        (command_id,AGENT,kind,Jsonb({"scheduled_action_id": SCHEDULE,"action":"suspend","ip_address":"127.0.0.1"})))
+        return command_id
+
+    def row(self, cid):
+        with connect() as conn, conn.cursor(row_factory=dict_row) as cur:
+            cur.execute("SELECT * FROM minerops_agent_command_queue WHERE id=%s", (cid,))
+            return dict(cur.fetchone())
+
+    def audits(self, cid):
+        with connect() as conn, conn.cursor(row_factory=dict_row) as cur:
+            cur.execute("""SELECT * FROM audit_logs WHERE action='agent.command.rejected'
+                           AND old_value_json->>'command_id'=%s""", (cid,))
+            return list(cur.fetchall())
+
+    def retire(self, path, cid):
+        if path == "poll": self.assertEqual(self.repo.claim_pending(AGENT), [])
+        elif path == "result":
+            with self.assertRaises(AgentCommandConflict): self.repo.finish_command(cid, AGENT, "completed", {}, None)
+        elif path == "cutover": return self.repo.retire_unsupported(AGENT, actor_user_id=ACTOR)
+        else: return self.repo.retire_schedule(SCHEDULE, actor_user_id=ACTOR, operation=path)
+
+    def assert_retired(self, cid, before, outcome):
+        after = self.row(cid)
+        self.assertEqual(after["command_status"], "cancelled")
+        self.assertIsNotNone(after["completed_at"])
+        for key in ("id","agent_id","command_type","claimed_at","created_at","request_payload","result_payload"):
+            self.assertEqual(after[key], before[key], key)
+        if before["error_message"]: self.assertTrue(after["error_message"].startswith(before["error_message"]+"\n"))
+        audits = self.audits(cid)
+        self.assertEqual(len(audits), 1)
+        self.assertEqual(audits[0]["old_value_json"]["status"], before["command_status"])
+        self.assertEqual(audits[0]["new_value_json"]["execution_outcome"], outcome)
+
+    def test_genuine_baseline_claim_then_explicit_repair_requeue_all_paths(self):
+        for path in PATHS:
+            with self.subTest(path=path):
+                self.clean(); cid=self.seed()
+                with connect() as conn, conn.cursor(row_factory=dict_row) as cur:
+                    cur.execute(BASELINE_CLAIM_SQL,(AGENT,5)); delivered=list(cur.fetchall())
+                self.assertEqual([r["command_id"] for r in delivered], [cid])
+                claimed=self.row(cid)
+                self.assertEqual(claimed["command_status"],"claimed"); self.assertIsNotNone(claimed["claimed_at"])
+                # Explicitly simulate a legacy repair. No production requeue path
+                # is asserted or added; retain the actual baseline claim evidence.
+                with connect() as conn, conn.cursor() as cur:
+                    cur.execute("UPDATE minerops_agent_command_queue SET command_status='pending' WHERE id=%s",(cid,))
+                before=self.row(cid); self.assertEqual(before["claimed_at"],claimed["claimed_at"])
+                self.retire(path,cid); self.assert_retired(cid,before,"unknown")
+                audit_before=self.audits(cid); row_before=self.row(cid)
+                for retry in PATHS: self.retire(retry,cid)
+                self.assertEqual(self.audits(cid),audit_before); self.assertEqual(self.row(cid),row_before)
+
+    def test_pristine_pending_all_paths_and_result_itself_is_delivery_evidence(self):
+        for path in PATHS:
+            with self.subTest(path=path):
+                self.clean(); cid=self.seed(); before=self.row(cid)
+                self.retire(path,cid)
+                self.assert_retired(cid,before,"unknown" if path=="result" else "not_dispatched")
+
+    def test_each_durable_evidence_source_is_unknown_on_every_path(self):
+        sources=("claim","completion","result","error","legacy_result","unlinked_legacy_result","audit_old","audit_new","prior_unknown")
+        for source in sources:
+            for path in PATHS:
+                with self.subTest(source=source,path=path):
+                    self.clean(); cid=self.seed()
+                    with connect() as conn, conn.cursor() as cur:
+                        if source in ("claim","completion","result","error"):
+                            assignments={"claim":"claimed_at=now()", "completion":"completed_at=now()",
+                                         "result":"result_payload='{}'::jsonb", "error":"error_message='ambiguous legacy error'"}
+                            cur.execute("UPDATE minerops_agent_command_queue SET "+assignments[source]+" WHERE id=%s",(cid,))
+                        elif source in ("legacy_result", "unlinked_legacy_result"):
+                            cur.execute("""INSERT INTO minerops_agent_command_results (agent_id,site_name,command_id,status,payload)
+                                           VALUES (%s,'simulated',%s,'completed','{}'::jsonb)""",(AGENT,cid if source=="legacy_result" else None))
+                        else:
+                            # Durable audit evidence with all queue timestamps missing.
+                            old={"command_id":cid,"agent_id":AGENT} if source!="audit_new" else {"agent_id":AGENT}
+                            new={"operation_id":cid} if source=="audit_new" else {"execution_outcome":"unknown"}
+                            action="agent.command.rejected" if source=="prior_unknown" else "legacy.command.delivered"
+                            cur.execute("""INSERT INTO audit_logs (actor_user_id,action,entity_type,old_value_json,new_value_json)
+                                           VALUES (%s,%s,'agent_command',%s,%s)""",(ACTOR,action,Jsonb(old),Jsonb(new)))
+                    prior_audits=self.audits(cid); before=self.row(cid); self.retire(path,cid)
+                    after=self.row(cid)
+                    self.assertEqual(after["command_status"],"cancelled")
+                    self.assertEqual(after["claimed_at"],before["claimed_at"])
+                    self.assertEqual(after["result_payload"],before["result_payload"])
+                    if before["completed_at"]: self.assertEqual(after["completed_at"],before["completed_at"])
+                    audits=self.audits(cid); self.assertEqual(len(audits),len(prior_audits)+1)
+                    self.assertTrue(all(a["new_value_json"].get("execution_outcome")=="unknown" for a in audits))
+
+    def test_creation_audit_alone_does_not_report_delivery(self):
+        cid=self.seed()
+        with connect() as conn, conn.cursor() as cur:
+            cur.execute("""INSERT INTO audit_logs (actor_user_id,action,entity_type,new_value_json)
+                           VALUES (%s,'miner.control.queued','miner',%s)""",(ACTOR,Jsonb({"operation_id":cid})))
+        before=self.row(cid); self.retire("cutover",cid); self.assert_retired(cid,before,"not_dispatched")
+
+    def test_currently_claimed_without_timestamp_remains_unknown(self):
+        for path in PATHS:
+            self.clean(); cid=self.seed()
+            with connect() as conn, conn.cursor() as cur:
+                cur.execute("UPDATE minerops_agent_command_queue SET command_status='claimed' WHERE id=%s",(cid,))
+            before=self.row(cid); self.retire(path,cid); self.assert_retired(cid,before,"unknown")
+
+    def test_audit_failure_rolls_back_every_cancellation_path(self):
+        for path in PATHS:
+            with self.subTest(path=path):
+                self.clean(); cid=self.seed(); before=self.row(cid)
+                def fail(cur,row,**kwargs): cur.execute("SELECT 1/0")
+                with patch.object(AgentCommandRepository,"_audit_rejection",side_effect=fail):
+                    with self.assertRaises(Exception): self.retire(path,cid)
+                self.assertEqual(self.row(cid),before); self.assertEqual(self.audits(cid),[])
+
+    def test_concurrent_baseline_claim_blocks_result_then_unknown(self):
+        cid=self.seed(); locker=connect()
+        try:
+            with locker.cursor() as cur:
+                cur.execute(BASELINE_CLAIM_SQL,(AGENT,1)); self.assertEqual(len(cur.fetchall()),1)
+            with ThreadPoolExecutor(max_workers=1) as pool:
+                def cancellation():
+                    def named_connection():
+                        conn=connect()
+                        conn.execute("SET application_name='stage-a-final-result-race'")
+                        return conn
+                    with self.assertRaises(AgentCommandConflict):
+                        AgentCommandRepository(named_connection).finish_command(cid,AGENT,"completed",{},None)
+                future=pool.submit(cancellation)
+                try:
+                    # Observe an actual database lock wait before committing the
+                    # claim, rather than relying on thread scheduling or sleeps.
+                    blocked=False; deadline=time.monotonic()+5
+                    while time.monotonic()<deadline:
+                        with connect() as observer, observer.cursor() as cur:
+                            cur.execute("""SELECT EXISTS (SELECT 1 FROM pg_stat_activity
+                                           WHERE application_name='stage-a-final-result-race'
+                                           AND wait_event_type='Lock')""")
+                            blocked=cur.fetchone()[0]
+                        if blocked: break
+                        time.sleep(.01)
+                    self.assertTrue(blocked,"Cancellation must wait for the genuine legacy claim lock")
+                finally:
+                    locker.commit()
+                future.result(timeout=10)
+            self.assertEqual(self.row(cid)["command_status"],"cancelled")
+            self.assertIsNotNone(self.row(cid)["claimed_at"])
+            self.assertEqual(self.audits(cid)[0]["new_value_json"]["execution_outcome"],"unknown")
+        finally: locker.close()
+
+    def test_mixed_legacy_backlog_classifies_each_row_without_delivering(self):
+        for n in range(12):
+            cid=self.seed("stage-a-final-evidence-"+str(n),"miner_control" if n%2 else "unknown_legacy")
+            if n%3:
+                with connect() as conn,conn.cursor() as cur:
+                    cur.execute("UPDATE minerops_agent_command_queue SET claimed_at=now() WHERE id=%s",(cid,))
+        with ThreadPoolExecutor(max_workers=2) as pool:
+            self.assertEqual(sum(pool.map(lambda _: self.repo.retire_unsupported(AGENT,actor_user_id=ACTOR), range(2))),12)
+        for n in range(12):
+            audit=self.audits("stage-a-final-evidence-"+str(n)); self.assertEqual(len(audit),1)
+            self.assertEqual(audit[0]["new_value_json"]["execution_outcome"],"unknown" if n%3 else "not_dispatched")

backend/tests/test_stage_a_schedule_cancellation_postgres.py

@@ -0,0 +1,309 @@
+"""Audited schedule retirement on an explicitly disposable PostgreSQL database."""
+
+import os
+import unittest
+from concurrent.futures import ThreadPoolExecutor
+from unittest.mock import patch
+from uuid import UUID
+
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+from psycopg.rows import dict_row
+from psycopg.types.json import Jsonb
+
+from app import main
+from app.agent_commands.exceptions import AgentCommandConflict
+from app.agent_commands.repository import AgentCommandRepository
+from app.agent_commands.policy import PHYSICAL_CONTROL_DISABLED
+from app.config import settings
+from app.identity import get_current_user, required_api_permission
+from tests.test_stage_a_containment_postgres import connect
+
+
+AGENT = "stage-a-schedule-agent"
+OTHER_AGENT = "stage-a-schedule-other-agent"
+ACTOR = UUID("00000000-0000-4000-8000-000000003301")
+SCHEDULE = "stage-a-schedule-history"
+
+
+class StageAScheduleCancellationTests(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        if os.environ.get("MINEROPS_STAGE_A_DISPOSABLE") != "1" or not settings.postgres_db.startswith("minerops_stage_a_"):
+            raise RuntimeError("Refusing schedule tests outside an explicitly disposable Stage A database")
+
+    def setUp(self):
+        self.no_transport = patch("socket.create_connection", side_effect=AssertionError("No miner connection"))
+        self.transport = self.no_transport.start()
+        self.addCleanup(self.no_transport.stop)
+        self.addCleanup(self.transport.assert_not_called)
+        with connect() as conn, conn.cursor() as cur:
+            cur.execute("DELETE FROM audit_logs WHERE actor_user_id = %s OR old_value_json ->> 'agent_id' IN (%s, %s)",
+                        (ACTOR, AGENT, OTHER_AGENT))
+            cur.execute("DELETE FROM minerops_agent_command_queue WHERE agent_id IN (%s, %s)", (AGENT, OTHER_AGENT))
+            cur.execute("DELETE FROM minerops_scheduled_bulk_actions WHERE id = %s", (SCHEDULE,))
+            cur.execute("""INSERT INTO users (id, username, display_name, password_hash, role_id)
+                           SELECT %s, 'stage-a-retirement-auditor', 'Retirement Auditor', 'unused', id
+                           FROM roles WHERE name = 'Admin' ON CONFLICT (id) DO NOTHING""", (ACTOR,))
+            cur.execute("""INSERT INTO minerops_scheduled_bulk_actions
+                           (id, agent_id, name, target_type, scheduled_for, status, last_error)
+                           VALUES (%s, %s, 'Historical schedule', 'all', now(), 'running', 'prior evidence')""",
+                        (SCHEDULE, AGENT))
+        self.repository = AgentCommandRepository(connect)
+
+    def seed(self, command_id, status="pending", agent=AGENT, kind="miner_control", schedule=SCHEDULE):
+        with connect() as conn, conn.cursor() as cur:
+            cur.execute("""INSERT INTO minerops_agent_command_queue
+                           (id, agent_id, command_type, command_status, request_payload, result_payload,
+                            error_message, claimed_at)
+                           VALUES (%s, %s, %s, %s, %s, %s, 'prior evidence',
+                                   CASE WHEN %s = 'claimed' THEN now() ELSE NULL END)""",
+                        (command_id, agent, kind, status,
+                         Jsonb({"scheduled_action_id": schedule, "ip_address": "192.0.2.1", "action": "suspend"}),
+                         Jsonb({"historical_result": True}), status))
+
+    @staticmethod
+    def row(command_id):
+        with connect() as conn, conn.cursor(row_factory=dict_row) as cur:
+            cur.execute("SELECT * FROM minerops_agent_command_queue WHERE id = %s", (command_id,))
+            return dict(cur.fetchone())
+
+    @staticmethod
+    def schedule():
+        with connect() as conn, conn.cursor(row_factory=dict_row) as cur:
+            cur.execute("SELECT * FROM minerops_scheduled_bulk_actions WHERE id = %s", (SCHEDULE,))
+            return dict(cur.fetchone())
+
+    @staticmethod
+    def audits():
+        with connect() as conn, conn.cursor(row_factory=dict_row) as cur:
+            cur.execute("""SELECT * FROM audit_logs WHERE actor_user_id = %s
+                           OR old_value_json ->> 'agent_id' IN (%s, %s)""", (ACTOR, AGENT, OTHER_AGENT))
+            return [dict(row) for row in cur.fetchall()]
+
+    def retire(self, operation="cancel"):
+        with patch.object(main, "get_db_connection", side_effect=connect):
+            return getattr(main, f"{operation}_dashboard_scheduled_bulk_action")(
+                SCHEDULE, {"id": str(ACTOR), "username": "retirement-auditor"},
+            )
+
+    def assert_retired(self, command_id, before, operation):
+        after = self.row(command_id)
+        self.assertEqual(after["command_status"], "cancelled")
+        for field in ("id", "agent_id", "command_type", "request_payload", "result_payload", "created_at", "claimed_at"):
+            self.assertEqual(after[field], before[field], field)
+        self.assertEqual(after["error_message"], "prior evidence\n" + PHYSICAL_CONTROL_DISABLED)
+        self.assertIsNotNone(after["completed_at"])
+        audits = [r for r in self.audits() if r["action"] == "agent.command.rejected"
+                  and r["old_value_json"]["command_id"] == command_id]
+        self.assertEqual(len(audits), 1)
+        audit = audits[0]
+        self.assertEqual(audit["actor_user_id"], ACTOR)
+        self.assertEqual(audit["old_value_json"]["agent_id"], AGENT)
+        self.assertEqual(audit["old_value_json"]["status"], before["command_status"])
+        self.assertEqual(audit["new_value_json"]["source"], "schedule_" + operation)
+        self.assertEqual(audit["new_value_json"]["reason"], PHYSICAL_CONTROL_DISABLED)
+        self.assertEqual(audit["new_value_json"]["execution_outcome"],
+                         "unknown")
+        self.assertNotIn("192.0.2.1", str(audit))
+
+    def test_schedule_cancel_retires_pending_and_claimed_with_attributed_audits(self):
+        for status in ("pending", "claimed"):
+            command_id = "stage-a-schedule-cancel-" + status
+            self.seed(command_id, status)
+        before = {s: self.row("stage-a-schedule-cancel-" + s) for s in ("pending", "claimed")}
+        self.retire()
+        for status in before:
+            self.assert_retired("stage-a-schedule-cancel-" + status, before[status], "cancel")
+        self.assertEqual(self.schedule()["status"], "cancelled")
+        self.assertEqual(len(self.audits()), 3)
+
+    def test_schedule_delete_is_audited_retirement_and_preserves_history(self):
+        self.seed("stage-a-schedule-delete", "claimed")
+        before, schedule_before = self.row("stage-a-schedule-delete"), self.schedule()
+        self.retire("delete")
+        self.assert_retired("stage-a-schedule-delete", before, "delete")
+        after = self.schedule()
+        for field in schedule_before:
+            if field not in {"status", "last_error", "updated_at"}:
+                self.assertEqual(after[field], schedule_before[field], field)
+        parent_audits = [r for r in self.audits() if r["action"] == "scheduled_action.retired"]
+        self.assertEqual(len(parent_audits), 1)
+        self.assertEqual(parent_audits[0]["new_value_json"]["operation"], "delete")
+
+    def test_repeated_cancel_and_delete_are_idempotent(self):
+        self.seed("stage-a-schedule-repeat")
+        self.retire()
+        before, schedule_before, audits_before = self.row("stage-a-schedule-repeat"), self.schedule(), self.audits()
+        for operation in ("cancel", "delete", "delete"):
+            self.retire(operation)
+        self.assertEqual(self.row("stage-a-schedule-repeat"), before)
+        self.assertEqual(self.schedule(), schedule_before)
+        self.assertEqual(self.audits(), audits_before)
+
+    def test_terminal_queue_and_schedule_history_remains_immutable(self):
+        for status in ("completed", "failed", "cancelled"):
+            self.seed("stage-a-schedule-terminal-" + status, status)
+        before = {s: self.row("stage-a-schedule-terminal-" + s) for s in ("completed", "failed", "cancelled")}
+        with connect() as conn, conn.cursor() as cur:
+            cur.execute("UPDATE minerops_scheduled_bulk_actions SET status = 'completed' WHERE id = %s", (SCHEDULE,))
+        schedule_before = self.schedule()
+        self.retire("delete")
+        self.assertEqual(self.schedule(), schedule_before)
+        for status, row in before.items():
+            self.assertEqual(self.row("stage-a-schedule-terminal-" + status), row)
+        self.assertEqual(self.audits(), [])
+
+    def test_stored_agent_and_schedule_scope_cannot_be_overridden(self):
+        self.seed("stage-a-schedule-own")
+        self.seed("stage-a-schedule-foreign", agent=OTHER_AGENT)
+        self.seed("stage-a-schedule-unrelated", schedule="another-schedule")
+        self.seed("stage-a-schedule-telemetry", kind="poll_known_miners_now")
+        protected = {c: self.row(c) for c in ("stage-a-schedule-foreign", "stage-a-schedule-unrelated", "stage-a-schedule-telemetry")}
+        self.retire()
+        self.assertEqual(self.row("stage-a-schedule-own")["command_status"], "cancelled")
+        for command_id, before in protected.items():
+            self.assertEqual(self.row(command_id), before)
+
+    def test_schedule_unknown_type_is_rejected_with_its_actual_reason(self):
+        self.seed("stage-a-schedule-unknown", kind="raw_api3")
+        self.retire()
+        rejection = next(a for a in self.audits() if a["action"] == "agent.command.rejected")
+        self.assertEqual(rejection["new_value_json"]["reason"], "unsupported_command_schema_stage_a")
+
+    def test_concurrent_cancel_delete_workers_audit_each_transition_once(self):
+        self.seed("stage-a-schedule-concurrent", "claimed")
+        def worker(operation):
+            return AgentCommandRepository(connect).retire_schedule(SCHEDULE, actor_user_id=str(ACTOR), operation=operation)
+        with ThreadPoolExecutor(max_workers=2) as pool:
+            counts = list(pool.map(worker, ("cancel", "delete")))
+        self.assertEqual(sorted(counts), [0, 1])
+        self.assertEqual(self.row("stage-a-schedule-concurrent")["command_status"], "cancelled")
+        self.assertEqual(len(self.audits()), 2)
+
+    def test_poll_finish_and_schedule_cancel_race_has_one_queue_audit(self):
+        self.seed("stage-a-schedule-race", "claimed")
+        def worker(i):
+            repo = AgentCommandRepository(connect)
+            try:
+                if i == 0:
+                    return repo.retire_schedule(SCHEDULE, actor_user_id=str(ACTOR), operation="cancel")
+                if i == 1:
+                    return repo.claim_pending(AGENT)
+                return repo.finish_command("stage-a-schedule-race", AGENT, "completed", {}, None)
+            except AgentCommandConflict:
+                return None
+        with ThreadPoolExecutor(max_workers=3) as pool:
+            list(pool.map(worker, range(3)))
+        rejections = [a for a in self.audits() if a["action"] == "agent.command.rejected"]
+        self.assertEqual(len(rejections), 1)
+        self.assertEqual(rejections[0]["new_value_json"]["execution_outcome"], "unknown")
+        self.assertEqual(self.row("stage-a-schedule-race")["command_status"], "cancelled")
+
+    def test_audit_failure_after_one_insert_rolls_back_all_cancellations(self):
+        for command_id in ("stage-a-schedule-rollback-a", "stage-a-schedule-rollback-b"):
+            self.seed(command_id, "claimed")
+        before = {c: self.row(c) for c in ("stage-a-schedule-rollback-a", "stage-a-schedule-rollback-b")}
+        schedule_before = self.schedule()
+        original = AgentCommandRepository._audit_rejection
+        calls = []
+        def fail_second(cur, row, **kwargs):
+            calls.append(row["id"])
+            if len(calls) == 2:
+                raise RuntimeError("simulated audit failure")
+            original(cur, row, **kwargs)
+        with patch.object(AgentCommandRepository, "_audit_rejection", side_effect=fail_second):
+            with self.assertRaisesRegex(RuntimeError, "simulated audit failure"):
+                self.retire("delete")
+        self.assertEqual(len(calls), 2)
+        self.assertEqual(self.schedule(), schedule_before)
+        for command_id, row in before.items():
+            self.assertEqual(self.row(command_id), row)
+        self.assertEqual(self.audits(), [])
+
+    def test_schedule_parent_audit_failure_rolls_back_queue_and_parent(self):
+        self.seed("stage-a-schedule-parent-rollback")
+        before, parent_before = self.row("stage-a-schedule-parent-rollback"), self.schedule()
+        # A database constraint failure after child audits exercises the final
+        # schedule/audit portion, rather than only an early application exception.
+        def connection_with_failure():
+            conn = connect()
+            with conn.cursor() as cur:
+                cur.execute("""CREATE OR REPLACE FUNCTION pg_temp.reject_schedule_audit() RETURNS trigger
+                               LANGUAGE plpgsql AS $$ BEGIN
+                               IF NEW.action = 'scheduled_action.retired' THEN RAISE EXCEPTION 'simulated parent audit failure'; END IF;
+                               RETURN NEW; END $$""")
+                cur.execute("CREATE TRIGGER stage_a_parent_audit_failure BEFORE INSERT ON audit_logs FOR EACH ROW EXECUTE FUNCTION pg_temp.reject_schedule_audit()")
+            # Trigger creation is in the very transaction the repository must roll back.
+            return conn
+        with self.assertRaisesRegex(Exception, "simulated parent audit failure"):
+            AgentCommandRepository(connection_with_failure).retire_schedule(
+                SCHEDULE, actor_user_id=str(ACTOR), operation="cancel")
+        self.assertEqual(self.row("stage-a-schedule-parent-rollback"), before)
+        self.assertEqual(self.schedule(), parent_before)
+        self.assertEqual(self.audits(), [])
+
+    def test_unknown_schedule_has_no_queue_or_audit_side_effects(self):
+        self.seed("stage-a-schedule-orphan")
+        before = self.row("stage-a-schedule-orphan")
+        with connect() as conn, conn.cursor() as cur:
+            cur.execute("DELETE FROM minerops_scheduled_bulk_actions WHERE id = %s", (SCHEDULE,))
+        from fastapi import HTTPException
+        with self.assertRaises(HTTPException) as error:
+            self.retire()
+        self.assertEqual(error.exception.status_code, 404)
+        self.assertEqual(self.row("stage-a-schedule-orphan"), before)
+        self.assertEqual(self.audits(), [])
+
+    def test_http_cancel_delete_require_control_permission(self):
+        app = FastAPI()
+        for operation in ("cancel", "delete"):
+            path = f"/api/dashboard/scheduled-bulk-actions/{{action_id}}/{operation}"
+            app.add_api_route(path, getattr(main, f"{operation}_dashboard_scheduled_bulk_action"), methods=["POST"])
+            self.assertEqual(required_api_permission("POST", path), "miners.control")
+        app.dependency_overrides[get_current_user] = lambda: {"id": str(ACTOR), "role": "Custom", "permissions": ["agents.manage"]}
+        with TestClient(app) as client, patch.object(main, "get_db_connection") as db:
+            for operation in ("cancel", "delete"):
+                self.assertEqual(client.post(f"/api/dashboard/scheduled-bulk-actions/{SCHEDULE}/{operation}").status_code, 403)
+            db.assert_not_called()
+
+    def test_offline_cutover_batch_is_audited_and_never_claims_readers(self):
+        with connect() as conn, conn.cursor() as cur:
+            cur.execute("""INSERT INTO minerops_agent_command_queue
+                           (id, agent_id, command_type, command_status, request_payload, claimed_at)
+                           SELECT 'stage-a-schedule-backlog-' || n, %s,
+                                  CASE WHEN n %% 2 = 0 THEN 'miner_control' ELSE 'raw_api3' END,
+                                  CASE WHEN n %% 3 = 0 THEN 'claimed' ELSE 'pending' END,
+                                  '{}'::jsonb, CASE WHEN n %% 5 = 0 THEN now() ELSE NULL END
+                           FROM generate_series(1,1003) n""", (AGENT,))
+        self.seed("stage-a-schedule-offline-read", kind="poll_known_miners_now")
+        self.assertEqual(self.repository.retire_unsupported(AGENT, actor_user_id=str(ACTOR)), 1000)
+        self.assertEqual(self.repository.retire_unsupported(AGENT, actor_user_id=str(ACTOR)), 3)
+        self.assertEqual(self.repository.retire_unsupported(AGENT, actor_user_id=str(ACTOR)), 0)
+        self.assertEqual(self.row("stage-a-schedule-offline-read")["command_status"], "pending")
+        rejections = self.audits()
+        self.assertEqual(len(rejections), 1003)
+        for audit in rejections:
+            self.assertEqual(audit["actor_user_id"], ACTOR)
+            self.assertEqual(audit["new_value_json"]["source"], "stage_a_cutover")
+            n = int(audit["old_value_json"]["command_id"].rsplit("-", 1)[1])
+            self.assertEqual(audit["new_value_json"]["execution_outcome"],
+                             "unknown" if n % 3 == 0 or n % 5 == 0 else "not_dispatched")
+
+    def test_cutover_zero_batch_does_not_claim_a_locked_drain_is_complete(self):
+        self.seed("stage-a-schedule-cutover-locked", "claimed")
+        locker = connect()
+        try:
+            with locker.cursor() as cur:
+                cur.execute("SELECT id FROM minerops_agent_command_queue WHERE id = %s FOR UPDATE", ("stage-a-schedule-cutover-locked",))
+                self.assertEqual(self.repository.retire_unsupported(AGENT, actor_user_id=str(ACTOR)), 0)
+                self.assertEqual(self.row("stage-a-schedule-cutover-locked")["command_status"], "claimed")
+        finally:
+            locker.rollback()
+            locker.close()
+        self.assertEqual(self.repository.retire_unsupported(AGENT, actor_user_id=str(ACTOR)), 1)
+        self.assertEqual(self.row("stage-a-schedule-cutover-locked")["command_status"], "cancelled")
+
+
+if __name__ == "__main__":
+    unittest.main()

docs/operations/STAGE_A_OLD_AGENT_CUTOVER.txt

@@ -0,0 +1,248 @@
+MinerOps Blocker #3 Stage A: old-Agent containment and upgrade runbook
+Status: review candidate. This procedure has NOT been executed in production.
+Scope: contain physical control; no Stages B-E protocol or control re-enablement.
+
+The independent audit demonstrated that an old Agent can execute a previously
+delivered command after the Hub marks it cancelled. Cancelling a database row,
+revoking a Hub credential, advertising control unavailable, or replacing script
+files does not revoke a cached command or loaded PowerShell functions. Execution
+of any previously claimed command remains UNKNOWN until separately investigated.
+
+Production approval requires independent security revalidation, an approved change
+window, verified network enforcement, and the native Windows checks below. A new
+database migration or production network infrastructure change requires explicit
+approval. This runbook authorizes no production actions by itself.
+
+Ownership and evidence
+
+Assign an incident/change owner, a network operator, a Hub/database operator, and a
+Windows operator. Record their identities, UTC timestamps, change ticket, and every
+gate's evidence in a protected location. Keep enrollment tokens and Agent/miner
+credentials out of ordinary logs and this checklist. Preserve required backups
+and recovery state with their existing ACLs; do not overwrite agent-config.json.
+
+Maintain one ledger containing ALL registered Agents, every distinct queue Agent
+(including deleted registry entries), offline/unreachable Agents, alternative
+installations and manually running executors, legacy shared identities, all Hub
+replicas/workers, scheduled jobs, task hosts, physical networks, and IPv4/IPv6 paths.
+For each Agent record its Site/Zone assignment, last check-in, host/network identity,
+network fence evidence, process/task evidence, installed-file hashes, verified
+artifact identifier, and release decision. Keep unverified entries quarantined.
+
+Gate 1: restrict physical write access BEFORE stopping/upgrading anything
+
+1. Arrange an approved maintenance window and identify the actual network control
+   that can keep old Agents from reaching ALL physical miners. Layer-4 rules cannot
+   distinguish a legacy read from a write on the same miner API port. Default to
+   denying Agent-to-miner traffic, including ports 4028 and 4433, alternative miner
+   management interfaces/ports, IPv6, alternate NICs/VPNs, and established sessions.
+   Telemetry may be interrupted. Do not substitute an unreviewed 'read-only' proxy.
+2. Apply only separately approved network controls through the responsible operator.
+   A stopped process or a host-local rule alone is insufficient for an unreachable
+   machine that can later reboot/reconnect. Use enforcement that already applies
+   to that machine's traffic when it returns. Include unknown/retired installations
+   and aliases; trusting an IP address alone is insufficient when addresses can change.
+3. Verify deny counters and session termination/conntrack handling. Use approved
+   synthetic targets/canaries, never a real write command to a physical miner.
+   A rule that blocks only NEW connections leaves a pre-existing API session exposed.
+   Check both the intended route and alternate routes. Keep the fence if evidence
+   is unavailable, ambiguous, or an offline Agent cannot be located.
+
+Exit gate: verified deny enforcement covers every unverified/legacy executor,
+including future reconnects and existing sessions. Failure means STOP; keep fences.
+
+Gate 2: stop old executors and automatic restarts
+
+1. Stop command production and processing on every old Hub/worker version, including
+   old schedule materializers and running requests that could still insert/deliver
+   commands. Hold ingress and scale/restart mechanisms as necessary under the
+   approved change procedure. Inventory the actual versions of every instance.
+2. On each reachable Windows host disable/hold all relevant Scheduled Task triggers,
+   retry/restart policies, service recovery, watchdogs, startup/logon entries, GUI
+   launch actions, manual jobs, and external orchestration that can launch MinerOps.
+   Record original definitions and intended running/stopped state for recovery.
+3. Stop MinerOpsAgent and any alternative/manual executor, then verify task instances
+   AND their PowerShell/child process trees have exited. Record command paths,
+   process start times and identities. An in-memory old module is still executable
+   even after its source file changes. Do not gracefully 'drain' by executing writes.
+4. Keep pool/enrichment tasks held during runtime replacement and verify no alternate
+   old code copy can auto-start. Never run uninstall or erase config/state to stop it.
+
+Exit gate: every reachable old executor is stopped and restart-held; every
+unreachable/unverified Agent remains network-fenced. Do not remove a fence to
+diagnose a failed stop. Escalate failures with evidence and keep maintenance holds.
+
+Gate 3: update ALL Hub instances with no mixed-version execution
+
+1. Capture protected queue/audit evidence and the relevant schedule/Agent inventory
+   before changes. Retain original request/result payloads, timestamps and errors.
+   Secure these exports because historical payloads may contain sensitive data.
+2. Install only the independently reviewed Stage A artifact on every Hub instance
+   while executor fences/holds remain. Stop old replicas and workers fully, including
+   rolling-update stragglers and old connections/requests; do not allow a mixture of
+   active baseline and Stage A schedulers/queue consumers.
+3. Verify each instance's actual artifact hashes and that physical admission is
+   fixed disabled: legacy write/diagnostic POSTs return 410 after authorization,
+   canonical typed controls return 409 physical_control_disabled_stage_a, and queue
+   SQL dispatches only the fixed read types. Do not use runtime flags to enable writes.
+4. Run reviewed schema compatibility checks as appropriate. Stage A remediation adds
+   NO migration. Stop and request approval if the deployed schema needs one.
+
+Exit gate: only reviewed containment Hub code is active; all old producers are
+stopped. Keep Agents held. If any instance cannot be verified, keep ingress/worker
+holds and physical fences rather than continuing a partial rolling deployment.
+
+Gate 4: audited cancellation across ALL Agents, including offline/queue-only IDs
+
+1. Use the canonical reviewed code and a real accountable operator users.id for
+   manual cutover retirement. Existing authenticated Agent polling/result rejection
+   is system-attributed by source and Agent ID; manual cutover supplies actor_user_id.
+   Do not impersonate another user, create invented execution evidence, or log secrets.
+2. Inventory IDs using the UNION of registry and queue Agent IDs. Also reconcile the
+   fleet ledger, historical/retired identities, and old shared Agent ID. Do not use
+   only Agents currently checking in. A useful read-only inventory is:
+
+     SELECT agent_id FROM minerops_site_agents
+     UNION SELECT agent_id FROM minerops_agent_command_queue
+     ORDER BY agent_id;
+
+3. In a separately approved administrative process configured for the intended DB,
+   call the reviewed repository operation for each ID, in committed bounded batches:
+
+     repository = AgentCommandRepository()
+     count = repository.retire_unsupported(agent_id, actor_user_id=operator_user_id)
+
+   Repeat for all IDs while progress is made, under an explicit time/batch limit.
+   This method never claims read work; do NOT substitute claim_pending or an Agent
+   HTTP poll as an administrative drain. Each transition and rejection audit commits
+   atomically. Database/audit failure rolls back and keeps the change window closed.
+4. A batch is bounded at 1000 and uses SKIP LOCKED. A zero count may mean locked rows,
+   not completion. Resolve/retry approved lock holders with time limits; never
+   silently declare the drain finished. Query ALL Agents to verify a zero live count:
+
+     SELECT agent_id, command_type, command_status, count(*)
+     FROM minerops_agent_command_queue
+     WHERE command_status IN ('pending', 'claimed')
+       AND command_type NOT IN
+           ('poll_known_miners_now', 'discover_new_miners', 'enrich_match_discovery')
+     GROUP BY agent_id, command_type, command_status;
+
+5. Reconcile newly cancelled rows to agent.command.rejected records. Check command ID,
+   Agent ID, prior state, immutable rejection reason, source and operator attribution.
+   Current pending status does not establish that a command was never delivered.
+   Current claimed status, any retained claimed_at, completion/result/error evidence,
+   linked legacy result or non-creation audit history, an unlinked legacy result
+   for that Agent, or an Agent result submission
+   requires execution_outcome=unknown. Missing or ambiguous evidence is unknown.
+   Only a pending row with a complete pre-cancellation check finding no delivery
+   evidence is classified not_dispatched by the queue. Historical claim evidence
+   survives cancellation; an explicitly repaired/requeued pending row is still
+   unknown. No ordinary production requeue mechanism is introduced by Stage A.
+   Neither classification proves absence of out-of-band historical physical writes.
+   Repeated retirement must add no duplicate audit and must not rewrite history.
+6. Schedule cancel/delete uses the stored schedule's Agent ID and the same rejection
+   audit policy; delete retains the schedule. This is NOT a fleet drain: orphaned or
+   foreign-Agent rows are handled by the per-Agent cutover sweep above. Never delete
+   queue/audit rows, reset claimed commands to pending, or rewrite terminal results.
+   Existing pre-remediation cancelled rows missing an audit must be preserved and
+   documented as a historical evidence gap. Do not invent a prior-state audit.
+
+Exit gate: the global live-unsupported query returns no rows, locks are accounted
+for, transition counts/audits reconcile, and no old producer can replenish work.
+This gate does not revoke already delivered payloads; gates 1, 2 and 5 remain mandatory.
+
+Gate 5: upgrade Windows code and verify installed bytes while stopped
+
+1. Use a verified artifact prepared from the exact independently reviewed Stage A
+   boundary. Record its digest and a trusted expected SHA-256 manifest outside the
+   target installation. VERSION is unchanged at 1.0.0-alpha and is not upgrade proof.
+2. Include all managed runtime scripts, especially MinerOpsAgentService.ps1,
+   MinerOpsControl.ps1, EnrichOneMiner.ps1, MinerOpsApi3EnrichmentPoller.ps1, the pool
+   pollers, task/GUI/tray scripts, and VERSION. Check the package's installer and
+   enrollment/credential-upgrade entry points as well. Include the manifest's origin
+   and review approval; hashes computed only from a suspect host are not a trust root.
+3. Confirm the main task was positively stopped BEFORE invoking the existing
+   installer. It can auto-restart a task that it found running. Keep the external
+   network fence and restart hold throughout installation and failure recovery.
+   Do not pass filesystem trust-reset switches unless their existing reviewed
+   recovery workflow specifically calls for them. Preserve config/state/logs/backups.
+4. Compare installed Get-FileHash -Algorithm SHA256 values to the trusted manifest.
+   Review task actions for the expected executable, canonical script path, arguments,
+   principal, and original/reviewed definition; check all aliases/duplicate tasks.
+   Verify ACLs, signer/source provenance where used, absence of old running processes,
+   and the task state consistent with the maintenance hold. Do not overwrite an
+   established task definition merely to force a convenient state.
+5. Follow the existing credential transition/enrollment recovery workflow only if
+   required. Hub credentials and local miner credentials are different trust domains;
+   rotating a Hub secret cannot recall a cached physical command. Keep credentials
+   local and protected. Never restore an old credential/config to make a test pass.
+
+Exit gate: installed bytes and process/task state match the approved ledger for
+every Agent proposed for release. An inaccessible Agent, hash mismatch, duplicate
+task, unexpected process, or trust-reset blocker remains fenced and stopped.
+
+Gate 6: native Windows validation and fresh process startup
+
+Before production release, run on a network-isolated native Windows PowerShell 5.1
+host with simulated miners only. Linux PowerShell 7 tests are not evidence of SYSTEM
+task lifecycle correctness. Validate these points and retain results:
+ - Actual SYSTEM/Scheduled Task stop, child cleanup, restart hold and reboot behavior.
+ - Installed package hash verification and fresh module loading after upgrade.
+ - Legacy/malformed/unsupported cached commands rejected with zero physical writes.
+ - Diagnostic write/parameter overrides rejected before any device connection;
+   fixed telemetry/discovery/API3 read selections remain compatible with simulators.
+ - Enrollment, credential-transition interruption/recovery, and running/stopped intent
+   retain existing protections, without credentials appearing in normal output.
+ - A full network fence prevents an old cached command from reaching a live simulator;
+   a separate unfenced positive control proves the simulator and old-executor setup.
+
+1. Start only a fresh verified Agent process, initially still fenced, after all
+   preceding gates are satisfied. Verify executable/script paths, start time after
+   publication, loaded source provenance, per-Agent authentication, Zone assignment,
+   and unavailable control advertisement. The advertisement alone is not proof.
+2. Recheck the unsupported queue and audit reconciliation after startup. Do not
+   restore read connectivity for a host with any unverified old executor.
+3. Release the fence only for individually verified upgraded hosts under the approved
+   network change. Re-enable only reviewed telemetry schedules. Physical control
+   stays disabled on Hub AND Agent. Retain independent monitor/rollback evidence.
+
+Exit gate: every released Agent has complete positive verification. Fleet-wide
+containment cannot be claimed while unverified devices have an unfenced route.
+
+Gate 7: returning offline Agents and rollback/recovery
+
+Maintain a default quarantine for every offline, unknown, historical or unverified
+Agent when it reconnects. A timestamp, IP address, same VERSION, successful Hub
+authentication, or control_available=false claim cannot release it. Repeat stop,
+hash/task verification, upgrade, fresh-start and relevant gates before restoring
+miner connectivity. Keep retired Agent credentials retired; do not reopen legacy
+shared-secret fallback for convenience.
+
+If ANY gate fails:
+ - Keep or reinstate physical network fencing immediately. Stop/restart-hold Agent
+   executors and prevent old Hub producers from running. Preserve evidence and notify
+   the change owner; do not automatically resume the previous running state.
+ - Database/audit failure: leave rolled-back rows as evidence, resolve the error,
+   retry audited retirement, and re-run the global zero-live check. Never repair
+   audit integrity by deleting history or fabricating execution outcomes.
+ - Installer/recovery failure: preserve trust blockers, recovery tokens and protected
+   backups according to existing recovery instructions. Verify state before retry.
+ - If restoring old Hub or Agent binaries is unavoidable, those binaries stay stopped
+   and all affected miner routes stay fenced. An operational rollback does not restore
+   security containment. Do not reconnect until the reviewed Stage A bytes are again
+   installed and every failed gate is repeated successfully.
+ - If verification cannot be completed within the change window, leave the affected
+   host quarantined with a named owner. Partial success is not authority to un-fence it.
+
+Disposable simulation available for independent review
+
+agents/windows/tests/Test-MinerOpsOldAgentFence.ps1 loads AST functions from a
+separate git archive of 2d89e80 and the audit's cached suspend payload. It requires
+MINEROPS_STAGE_A_SIMULATION=1 and an explicitly provided simulated target. The
+remediation artifact directory contains simulate_fence.py and simulated_miner.py
+which create a temporary Docker INTERNAL network and fake API3 miner, then run the
+old executor first with --network none (zero simulated writes) and separately on
+the internal network (one simulated service-stop write). No host ports are exposed.
+Only those created resources are removed. This demonstrates the coarse fence's
+effect, not the adequacy of any production firewall/NAC design. Native Windows
+evidence and actual fleet/network verification remain separate release gates.

scripts/test-miner-control.sh

@@ -16,7 +16,7 @@ cleanup() {
 }
 trap cleanup EXIT
 [[ -f .env.dev ]] || { echo "Missing .env.dev" >&2; exit 1; }
-for source_file in backend/tests/test_agent_auth.py backend/tests/test_miner_control.py backend/tests/test_agent_commands.py backend/tests/test_miner_control_postgres.py; do
+for source_file in backend/tests/test_agent_auth.py backend/tests/test_miner_control.py backend/tests/test_agent_commands.py backend/tests/test_miner_control_postgres.py backend/tests/test_legacy_control_retirement.py; do
     permissions="$(stat -c '%a' "${source_file}")"
     (( (10#${permissions} % 10) & 4 )) || {
         echo "${source_file} is not readable by the non-root test container" >&2
@@ -47,6 +47,7 @@ done
         tests.test_agent_auth \
         tests.test_miner_control \
         tests.test_agent_commands \
+        tests.test_legacy_control_retirement \
         tests.test_miner_control_postgres
 
 echo "Miner Control Foundation focused tests passed."

scripts/test-stage-a-security.sh

@@ -0,0 +1,28 @@
+#!/usr/bin/env bash
+# Run inside an isolated test environment with a freshly migrated disposable DB.
+# No service creation, cleanup, miner commands, or production defaults.
+set -euo pipefail
+repo_root="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
+cd "${repo_root}/backend"
+[[ "${MINEROPS_STAGE_A_DISPOSABLE:-}" == 1 ]] || {
+    echo "Require MINEROPS_STAGE_A_DISPOSABLE=1 and an isolated test database." >&2
+    exit 1
+}
+export PYTHONDONTWRITEBYTECODE=1
+export MINEROPS_ENABLE_REAL_REBOOT=false
+export MINEROPS_ENABLE_REAL_POWER_CONTROL=false
+python -c 'from app.config import settings; assert settings.postgres_db.startswith("minerops_stage_a_"), "Refusing non-Stage-A database"'
+python -m unittest -v \
+    tests.test_legacy_control_retirement \
+    tests.test_stage_a_containment \
+    tests.test_stage_a_containment_postgres \
+    tests.test_stage_a_delivery_evidence \
+    tests.test_stage_a_delivery_evidence_postgres \
+    tests.test_stage_a_schedule_cancellation_postgres \
+    tests.test_agent_zones_postgres \
+    tests.test_miner_control \
+    tests.test_agent_commands.AgentCommandServiceTests \
+    tests.test_agent_commands.AgentCommandRouterTests
+# Separately run agents/windows/tests/Test-MinerOpsControl.ps1 in a network-free
+# PowerShell environment, then Test-MinerOpsDiagnosticTransport.ps1. Both load
+# functions only. Diagnostic compatibility listeners bind loopback exclusively.