#!/usr/bin/env python3
"""Exact SSH management/lifecycle/close/slot-selection functions, deterministic RTOS.
No wolfSSH, sockets, real scheduling or target execution is claimed.
"""
from pathlib import Path
import re
import subprocess
import tempfile

ROOT = Path(__file__).resolve().parents[2]
source = (ROOT / 'src/ssh_transport.c').read_text()
def function(name):
    match = re.search(r'^(?:static )?[^\n]+\b' + name + r'\([^;]*?\n\{.*?^\}', source, re.M | re.S)
    assert match, name
    return match.group() + '\n'
header = '\n'.join(line for line in (ROOT / 'src/ssh_transport.h').read_text().splitlines() if not line.startswith(('#include', '#pragma once')))
constants = '\n'.join(re.search(r'^#define ' + name + r' .+$', source, re.M).group() for name in ('SSH_TRANSPORT_GENERATION_MAX', 'SSH_TRANSPORT_COMMAND_TIMEOUT_MS', 'SSH_TRANSPORT_MAX_PENDING_HANDSHAKES'))
username = re.search(r'^#define USER_DATABASE_USERNAME_CAPACITY .+$', (ROOT / 'src/user_database.h').read_text(), re.M).group()
fakes = r'''
#include <assert.h>
#include <stdbool.h>
#include <stdint.h>
#include <stddef.h>
#include <string.h>
#include <stdio.h>
typedef int esp_err_t;
enum { ESP_OK, ESP_FAIL, ESP_ERR_INVALID_ARG, ESP_ERR_INVALID_STATE, ESP_ERR_TIMEOUT, ESP_ERR_NOT_FOUND };
typedef uint32_t session_broker_client_id_t;
typedef int user_role_t;
typedef int user_auth_method_t;
'''
state = r'''
typedef struct { ssh_transport_session_state_t state; uint32_t generation, session_id; } ssh_slot_t;
static ssh_slot_t s_slots[SSH_TRANSPORT_MAX_SESSIONS];
static ssh_transport_session_snapshot_t s_session_snapshots[SSH_TRANSPORT_MAX_SESSIONS];
static uint32_t s_external_close_id[SSH_TRANSPORT_MAX_SESSIONS];
static unsigned depth, mutex_storage, notifications, ticks;
static unsigned *s_command_mutex = &mutex_storage;
static bool s_initialized, s_running, s_transitioning, s_cleanup_pending, s_desired_running;
static uint32_t s_management_generation, s_requested_sequence, s_completed_sequence;
static int s_command_result;
static bool owner_stalled, owner_fail;
static uint32_t identity_generation = 3, identity_token;
static unsigned replacements;
static bool persist_fail;
static esp_err_t ssh_security_reserve_identity(uint32_t generation, bool reset, uint32_t *token) {
    (void)reset; assert(mutex_storage && !depth); *token = 0;
    if (identity_token || (generation && generation != identity_generation)) return ESP_ERR_INVALID_STATE;
    *token = identity_token = 1; return ESP_OK;
}
static esp_err_t ssh_security_replace_reserved(uint32_t token) {
    assert(token == identity_token && mutex_storage && !depth && !s_running && !s_cleanup_pending);
    ++replacements; if (persist_fail) return ESP_FAIL; ++identity_generation; return ESP_OK;
}
static void ssh_security_release_identity(uint32_t token) { if (token) { assert(identity_token == token && mutex_storage && !depth); identity_token = 0; } }
static int s_lock;
#define taskENTER_CRITICAL(p) do { (void)(p); assert(depth++ == 0); } while(0)
#define taskEXIT_CRITICAL(p) do { (void)(p); assert(--depth == 0); } while(0)
#define pdTRUE 1
#define portMAX_DELAY 99999U
#define pdMS_TO_TICKS(n) (n)
typedef unsigned TickType_t;
static int xSemaphoreTake(unsigned *m, unsigned wait) { assert(!depth); (void)wait; if (*m) return 0; *m = 1; return 1; }
static void xSemaphoreGive(unsigned *m) { assert(!depth && *m); *m = 0; }
static void notify_task(void) { assert(!depth); ++notifications; }
static unsigned xTaskGetTickCount(void) { return ticks; }
static void vTaskDelay(unsigned n) {
    assert(!depth && mutex_storage); ticks += n;
    if (!owner_stalled) {
        s_completed_sequence = s_requested_sequence; s_command_result = owner_fail ? ESP_FAIL : ESP_OK;
        s_running = owner_fail ? false : s_desired_running; s_transitioning = false; s_cleanup_pending = owner_fail;
    }
}
'''
tests = r'''
static void reset(void) {
    memset(s_slots, 0, sizeof(s_slots)); memset(s_session_snapshots, 0, sizeof(s_session_snapshots));
    memset(s_external_close_id, 0, sizeof(s_external_close_id));
    s_initialized = s_running = true; s_transitioning = s_cleanup_pending = owner_stalled = owner_fail = false;
    mutex_storage = notifications = ticks = 0; s_management_generation = 7; s_requested_sequence = s_completed_sequence = 0;
    for (unsigned i = 0; i < 2; ++i) {
        s_slots[i] = (ssh_slot_t){SSH_TRANSPORT_SESSION_ACTIVE, 2, make_session_id(i, 2)};
        s_session_snapshots[i] = (ssh_transport_session_snapshot_t){.active=true, .session_id=s_slots[i].session_id, .generation=2, .state=SSH_TRANSPORT_SESSION_ACTIVE};
    }
}
int main(void) {
    reset(); bool committed = true;
    assert(ssh_transport_replace_identity(6,3,false,&committed)==ESP_ERR_INVALID_STATE && !committed && !notifications && !replacements);
    assert(ssh_transport_replace_identity(7,2,false,&committed)==ESP_ERR_INVALID_STATE && !notifications && !replacements);
    identity_token=1; assert(ssh_transport_replace_identity(7,3,false,&committed)==ESP_ERR_INVALID_STATE && !notifications); identity_token=0;
    mutex_storage=1; assert(ssh_transport_replace_identity(7,3,false,&committed)==ESP_ERR_TIMEOUT && !notifications); mutex_storage=0;
    owner_fail=true; assert(ssh_transport_replace_identity(7,3,false,&committed)==ESP_FAIL && !committed && !replacements && notifications==1 && s_cleanup_pending && !identity_token);
    assert(ssh_transport_start()==ESP_ERR_INVALID_STATE && notifications==1);
    reset(); owner_stalled=true;
    assert(ssh_transport_replace_identity(7,3,false,&committed)==ESP_ERR_TIMEOUT && !committed && !replacements && notifications==1 && s_transitioning && !identity_token);
    reset(); persist_fail=true;
    assert(ssh_transport_replace_identity(7,3,false,&committed)==ESP_FAIL && !committed && replacements==1 && notifications==2 && s_running && identity_generation==3);
    reset(); persist_fail=false;
    assert(ssh_transport_replace_identity(7,3,false,&committed)==ESP_OK && committed && s_running && identity_generation==4 && notifications==2);
    assert(ssh_transport_replace_identity(7,3,false,&committed)==ESP_ERR_INVALID_STATE && !committed && notifications==2);
    reset(); s_running=false;
    assert(ssh_transport_replace_identity(7,4,false,&committed)==ESP_OK && committed && !s_running && !notifications);
    assert(ssh_transport_replace_host_key(true)==ESP_OK && s_running && notifications==1);
    puts("PASS SSH combined generation/service-owner admission, competing reservation, failed-stop no mutation/start, timeout retention, persistence recovery, replay fence and stopped/reset semantics");
    reset(); ssh_transport_management_snapshot_t v;
    assert(ssh_transport_get_management_snapshot(NULL) == ESP_ERR_INVALID_ARG);
    assert(ssh_transport_get_management_snapshot(&v) == ESP_OK && v.generation == 7 && v.running && !v.transitioning);
    assert(!notifications && !depth && !mutex_storage);
    s_initialized = false; assert(ssh_transport_get_management_snapshot(&v) == ESP_ERR_INVALID_STATE); s_initialized = true;
    assert(ssh_transport_manage_current(SSH_TRANSPORT_MANAGE_DISCONNECT, 9, 7) == ESP_OK);
    assert(s_external_close_id[0] == 9 && !s_external_close_id[1] && notifications == 1);
    assert(ssh_transport_get_management_snapshot(&v) == ESP_OK && v.sessions[0].close_requested && !v.sessions[1].close_requested);
    assert(ssh_transport_manage_current(SSH_TRANSPORT_MANAGE_DISCONNECT, 9, 7) == ESP_ERR_NOT_FOUND);
    assert(consume_external_close(&s_slots[0], 0) && !s_external_close_id[0] && s_session_snapshots[0].close_requested);
    assert(!consume_external_close(&s_slots[1], 1));
    puts("PASS SSH atomic published snapshot/target close, duplicate rejection and unrelated-slot isolation");
    reset();
    s_external_close_id[0] = 5; assert(!consume_external_close(&s_slots[0], 0));
    s_slots[0].state = SSH_TRANSPORT_SESSION_FREE; assert(!consume_external_close(&s_slots[0], 0) && !s_external_close_id[0]);
    s_session_snapshots[0].active = false;
    assert(ssh_transport_manage_current(SSH_TRANSPORT_MANAGE_DISCONNECT, 9, 7) == ESP_ERR_NOT_FOUND);
    s_session_snapshots[0].active = true; s_session_snapshots[0].session_id = 13;
    assert(ssh_transport_manage_current(SSH_TRANSPORT_MANAGE_DISCONNECT, 9, 7) == ESP_ERR_NOT_FOUND);
    assert(!s_external_close_id[0] && !notifications);
    size_t index; s_slots[0].generation = SSH_TRANSPORT_GENERATION_MAX; s_slots[1].state = SSH_TRANSPORT_SESSION_FREE;
    assert(find_free_slot(&index) == &s_slots[1] && index == 1);
    s_slots[1].generation = SSH_TRANSPORT_GENERATION_MAX; assert(find_free_slot(&index) == NULL);
    assert(make_session_id(1, SSH_TRANSPORT_GENERATION_MAX) != 0);
    puts("PASS SSH disconnect/reuse/late owner close safety and generation exhaustion retires slots");
    reset();
    assert(ssh_transport_manage_current(SSH_TRANSPORT_MANAGE_STOP, 0, 6) == ESP_ERR_INVALID_STATE);
    assert(ssh_transport_manage_current(SSH_TRANSPORT_MANAGE_START, 0, 7) == ESP_ERR_INVALID_STATE);
    assert(ssh_transport_manage_current(SSH_TRANSPORT_MANAGE_STOP, 9, 7) == ESP_ERR_INVALID_ARG);
    assert(ssh_transport_manage_current(SSH_TRANSPORT_MANAGE_DISCONNECT, 0, 7) == ESP_ERR_INVALID_ARG);
    assert(ssh_transport_manage_current(99, 0, 7) == ESP_ERR_INVALID_ARG);
    mutex_storage = 1; assert(ssh_transport_manage_current(SSH_TRANSPORT_MANAGE_STOP, 0, 7) == ESP_ERR_TIMEOUT); mutex_storage = 0;
    s_transitioning = true; assert(ssh_transport_manage_current(SSH_TRANSPORT_MANAGE_DISCONNECT, 9, 7) == ESP_ERR_INVALID_STATE); s_transitioning = false;
    assert(!notifications);
    assert(ssh_transport_manage_current(SSH_TRANSPORT_MANAGE_STOP, 0, 7) == ESP_OK && !s_running && s_management_generation == 8);
    assert(ssh_transport_manage_current(SSH_TRANSPORT_MANAGE_START, 0, 7) == ESP_ERR_INVALID_STATE);
    assert(ssh_transport_manage_current(SSH_TRANSPORT_MANAGE_START, 0, 8) == ESP_OK && s_running && s_management_generation == 9);
    assert(ssh_transport_manage_current(SSH_TRANSPORT_MANAGE_DISCONNECT, 9, 7) == ESP_ERR_INVALID_STATE);
    assert(ssh_transport_stop() == ESP_OK && s_management_generation == 10);
    assert(ssh_transport_start() == ESP_OK && s_management_generation == 11);
    assert(ssh_transport_manage_current(SSH_TRANSPORT_MANAGE_STOP, 0, 9) == ESP_ERR_INVALID_STATE);
    puts("PASS SSH conditional start/stop, command mutex, CLI transitions and stop/start ABA fencing");
    reset(); owner_stalled = true;
    assert(ssh_transport_manage_current(SSH_TRANSPORT_MANAGE_STOP, 0, 7) == ESP_ERR_TIMEOUT && s_transitioning && !mutex_storage && s_management_generation == 8);
    assert(ssh_transport_manage_current(SSH_TRANSPORT_MANAGE_START, 0, 8) == ESP_ERR_INVALID_STATE);
    assert(ssh_transport_get_management_snapshot(&v) == ESP_OK && v.transitioning);
    reset(); owner_fail = true;
    assert(ssh_transport_manage_current(SSH_TRANSPORT_MANAGE_STOP, 0, 7) == ESP_FAIL && s_cleanup_pending);
    assert(ssh_transport_manage_current(SSH_TRANSPORT_MANAGE_START, 0, 8) == ESP_ERR_INVALID_STATE);
    owner_fail = false; assert(ssh_transport_stop() == ESP_OK && !s_cleanup_pending);
    s_management_generation = UINT32_MAX - 1;
    assert(ssh_transport_manage_current(SSH_TRANSPORT_MANAGE_START, 0, UINT32_MAX - 1) == ESP_OK && s_management_generation == UINT32_MAX);
    assert(ssh_transport_manage_current(SSH_TRANSPORT_MANAGE_STOP, 0, UINT32_MAX) == ESP_ERR_INVALID_ARG);
    assert(ssh_transport_stop() == ESP_OK && s_management_generation == UINT32_MAX);
    puts("PASS SSH admitted timeout is not cancellation; failed cleanup and saturated versions preserve CLI recovery");
}
'''
names = ('next_generation', 'make_session_id', 'consume_external_close', 'find_free_slot', 'request_running_locked', 'request_running', 'ssh_transport_start', 'ssh_transport_stop', 'ssh_transport_replace_identity', 'ssh_transport_replace_host_key', 'ssh_transport_get_management_snapshot', 'ssh_transport_manage_current')
# Guard the accept path, which is not executed with the socket double here.
assert 'uint32_t generation = slot->generation + 1U;' in function('accept_connections')
assert 'next_generation(slot->generation)' not in source
with tempfile.TemporaryDirectory(prefix='ssh-management-') as directory:
    tmp = Path(directory)
    unit = fakes + username + '\n' + header + '\n' + constants + '\n' + state + '\n'.join(function(n) for n in names) + tests
    (tmp / 'test.c').write_text(unit)
    subprocess.run(['cc', '-std=c11', '-Wall', '-Wextra', '-Werror', str(tmp / 'test.c'), '-o', str(tmp / 'test')], check=True, timeout=30)
    subprocess.run([str(tmp / 'test')], check=True, timeout=10)