story/scripts/load-conversation-state.ps1

param(
    [int]$ConversationCount=1000,
    [int]$ExpiredCount=50,
    [int]$TimeoutSeconds=30
)
$ErrorActionPreference='Stop'
$PSNativeCommandUseErrorActionPreference=$true
if($ConversationCount-lt1-or$ConversationCount-gt10000-or$ExpiredCount-lt1-or$ExpiredCount-gt$ConversationCount){throw 'Load counts are outside the supported safety boundary'}
$root=Split-Path -Parent $PSScriptRoot
Set-Location -LiteralPath $root
$run=[Guid]::NewGuid().ToString('N')
$model="load-state-$run"
$prefix="load-state-$run-"
$tenant='10000000-0000-0000-0000-000000000001'

function Sql([string]$statement){
    $result=$statement|docker compose exec -T postgres psql -U voice -d voice_platform -v ON_ERROR_STOP=1 -q -t -A
    if($LASTEXITCODE-ne0){throw 'Conversation state load SQL failed'}
    return ($result|Out-String).Trim()
}

try{
    $setup=@"
BEGIN;
CREATE TEMP TABLE load_base ON COMMIT DROP AS
WITH selected AS (
  SELECT household_id,id subject_id,tenant_id,profile_version FROM subject_profile WHERE tenant_id='$tenant'::uuid AND status='active' ORDER BY created_at LIMIT 1
), consent AS (
  INSERT INTO consent_snapshot(tenant_id,subject_id,snapshot_version,effective_consents,source_consent_ids)
  SELECT tenant_id,subject_id,profile_version,'{}'::jsonb,jsonb_build_array('$model') FROM selected
  RETURNING id,tenant_id,subject_id
), context AS (
  INSERT INTO context_snapshot(tenant_id,subject_id,profile_version,consent_snapshot_id,memory_versions,prompt_bundle_version,engine_mode,content_digest,sanitized_summary)
  SELECT c.tenant_id,c.subject_id,s.profile_version,c.id,'[]'::jsonb,'$model','s2s_hybrid',encode(digest('$model','sha256'),'hex'),'合成通话状态负载夹具'
  FROM consent c JOIN selected s ON s.subject_id=c.subject_id
  RETURNING id,tenant_id,subject_id,consent_snapshot_id
)
SELECT s.household_id,s.subject_id,c.consent_snapshot_id,c.id context_id FROM selected s JOIN context c ON c.subject_id=s.subject_id;
CREATE TEMP TABLE load_devices ON COMMIT DROP AS
WITH inserted AS (
  INSERT INTO device(tenant_id,serial_digest,model,hardware_revision,credential_ref,minimum_security_version)
  SELECT '$tenant'::uuid,encode(digest('$run-device-'||g,'sha256'),'hex'),'$model','load','load://state/'||g,'load'
  FROM generate_series(1,$ConversationCount) g
  RETURNING id
) SELECT id,row_number() OVER() AS n FROM inserted;
WITH inserted AS (
  INSERT INTO conversation(tenant_id,household_id,subject_id,device_id,consent_snapshot_id,context_snapshot_id,provider_task_ref,engine_mode,retention_mode,status,allocation_expires_at,lease_expires_at,started_at)
  SELECT '$tenant'::uuid,b.household_id,b.subject_id,d.id,b.consent_snapshot_id,b.context_id,'$prefix'||d.n,'s2s_hybrid','none','ACTIVE',now()+interval '10 minutes',now()+interval '10 minutes',now()
  FROM load_devices d CROSS JOIN load_base b RETURNING id,provider_task_ref
)
INSERT INTO rtc_task(conversation_id,provider,provider_task_ref,room_id,device_rtc_user_id,agent_rtc_user_id,status,lease_expires_at)
SELECT id,'mock',provider_task_ref,'room-'||provider_task_ref,'device-'||provider_task_ref,'agent-'||provider_task_ref,'ACTIVE',now()+interval '10 minutes' FROM inserted;
WITH targets AS (SELECT id FROM conversation WHERE provider_task_ref LIKE '$prefix%' ORDER BY provider_task_ref LIMIT $ExpiredCount)
UPDATE conversation SET lease_expires_at=now()-interval '1 minute',updated_at=now()-interval '1 minute' WHERE id IN(SELECT id FROM targets);
COMMIT;
SELECT count(*)||'|'||count(*) FILTER(WHERE lease_expires_at<now()) FROM conversation WHERE provider_task_ref LIKE '$prefix%';
"@
    $evidence=Sql $setup
    if($evidence-ne"$ConversationCount|$ExpiredCount"){throw "Conversation load fixture mismatch: $evidence"}
    $deadline=[DateTime]::UtcNow.AddSeconds($TimeoutSeconds)
    do{
        Start-Sleep -Milliseconds 500
        $state=Sql "SELECT count(*) FILTER(WHERE status='ACTIVE')||'|'||count(*) FILTER(WHERE status='STOPPING')||'|'||count(*) FROM conversation WHERE provider_task_ref LIKE '$prefix%';"
    }while($state-ne"$($ConversationCount-$ExpiredCount)|$ExpiredCount|$ConversationCount"-and[DateTime]::UtcNow-lt$deadline)
    if($state-ne"$($ConversationCount-$ExpiredCount)|$ExpiredCount|$ConversationCount"){throw "Recovery scan did not converge exactly: $state"}
    $recovery=Sql "SELECT count(*)||'|'||count(DISTINCT aggregate_id)||'|'||COALESCE(max(per_conversation),0) FROM (SELECT aggregate_id,count(*) per_conversation FROM outbox_event WHERE event_type='conversation.agent_stop_requested' AND aggregate_id IN(SELECT id FROM conversation WHERE provider_task_ref LIKE '$prefix%') GROUP BY aggregate_id) x;"
    if($recovery-ne"$ExpiredCount|$ExpiredCount|1"){throw "Recovery generated missing or duplicate stop intents: $recovery"}
    $wrong=Sql "SELECT count(*) FROM conversation WHERE provider_task_ref LIKE '$prefix%' AND ((lease_expires_at>=now() AND status<>'ACTIVE') OR (lease_expires_at<now() AND status<>'STOPPING'));"
    if($wrong-ne'0'){throw "Recovery incorrectly changed healthy or failed to change expired conversations: $wrong"}
    Write-Host "Conversation state load passed: total=$ConversationCount, healthy_active=$($ConversationCount-$ExpiredCount), expired_recovered=$ExpiredCount, stop_intents=$recovery, no_double_task=true"
}finally{
    $cleanup=@"
BEGIN;
CREATE TEMP TABLE load_conversations ON COMMIT DROP AS SELECT id,public_id FROM conversation WHERE provider_task_ref LIKE '$prefix%';
CREATE TEMP TABLE load_snapshots ON COMMIT DROP AS SELECT id context_id,consent_snapshot_id FROM context_snapshot WHERE prompt_bundle_version='$model';
DELETE FROM alert_incident WHERE tenant_id='$tenant'::uuid AND resource_type='conversation' AND resource_public_id IN(SELECT public_id FROM load_conversations);
DELETE FROM web_event WHERE resource_type='conversation' AND resource_public_id IN(SELECT public_id FROM load_conversations);
DELETE FROM audit_event WHERE resource_id IN(SELECT id FROM load_conversations);
DELETE FROM outbox_event WHERE aggregate_id IN(SELECT id FROM load_conversations);
DELETE FROM conversation_state_history WHERE conversation_id IN(SELECT id FROM load_conversations);
DELETE FROM rtc_task WHERE conversation_id IN(SELECT id FROM load_conversations);
DELETE FROM conversation WHERE id IN(SELECT id FROM load_conversations);
DELETE FROM device WHERE model='$model';
DELETE FROM context_snapshot WHERE id IN(SELECT context_id FROM load_snapshots);
DELETE FROM consent_snapshot WHERE id IN(SELECT consent_snapshot_id FROM load_snapshots);
COMMIT;
"@
    try{Sql $cleanup|Out-Null}catch{Write-Warning $_}
}