import * as SqlClient from "effect/unstable/sql/SqlClient"; import * as Effect from "effect/Effect"; export default Effect.gen(function* () { const sql = yield* SqlClient.SqlClient; yield* sql` INSERT OR IGNORE INTO projection_pending_approvals ( request_id, thread_id, turn_id, status, decision, created_at, resolved_at ) SELECT requested.request_id, requested.thread_id, requested.turn_id, 'pending', NULL, requested.created_at, NULL FROM ( SELECT json_extract(payload_json, '$.requestId') AS request_id, thread_id, turn_id, created_at, ROW_NUMBER() OVER ( PARTITION BY json_extract(payload_json, '$.requestId') ORDER BY created_at ASC, activity_id ASC ) AS row_number FROM projection_thread_activities WHERE kind = 'approval.requested' AND json_extract(payload_json, '$.requestId') IS NOT NULL ) AS requested WHERE requested.row_number = 1 `; yield* sql` WITH latest_resolutions AS ( SELECT resolved.request_id, resolved.resolved_at, resolved.decision FROM ( SELECT json_extract(payload_json, '$.requestId') AS request_id, created_at AS resolved_at, CASE WHEN json_extract(payload_json, '$.decision') IN ( 'accept', 'acceptForSession', 'decline', 'cancel' ) THEN json_extract(payload_json, '$.decision') ELSE NULL END AS decision, ROW_NUMBER() OVER ( PARTITION BY json_extract(payload_json, '$.requestId') ORDER BY created_at DESC, activity_id DESC ) AS row_number FROM projection_thread_activities WHERE kind = 'approval.resolved' AND json_extract(payload_json, '$.requestId') IS NOT NULL ) AS resolved WHERE resolved.row_number = 1 ) UPDATE projection_pending_approvals SET status = 'resolved', decision = ( SELECT latest_resolutions.decision FROM latest_resolutions WHERE latest_resolutions.request_id = projection_pending_approvals.request_id ), resolved_at = ( SELECT latest_resolutions.resolved_at FROM latest_resolutions WHERE latest_resolutions.request_id = projection_pending_approvals.request_id ) WHERE EXISTS ( SELECT 1 FROM latest_resolutions WHERE latest_resolutions.request_id = projection_pending_approvals.request_id ) `; yield* sql` WITH latest_response_events AS ( SELECT response.request_id, response.resolved_at, response.decision FROM ( SELECT json_extract(payload_json, '$.requestId') AS request_id, occurred_at AS resolved_at, CASE WHEN json_extract(payload_json, '$.decision') IN ( 'accept', 'acceptForSession', 'decline', 'cancel' ) THEN json_extract(payload_json, '$.decision') ELSE NULL END AS decision, ROW_NUMBER() OVER ( PARTITION BY json_extract(payload_json, '$.requestId') ORDER BY occurred_at DESC, sequence DESC ) AS row_number FROM orchestration_events WHERE event_type = 'thread.approval-response-requested' AND json_extract(payload_json, '$.requestId') IS NOT NULL ) AS response WHERE response.row_number = 1 ) UPDATE projection_pending_approvals SET status = 'resolved', decision = ( SELECT latest_response_events.decision FROM latest_response_events WHERE latest_response_events.request_id = projection_pending_approvals.request_id ), resolved_at = ( SELECT latest_response_events.resolved_at FROM latest_response_events WHERE latest_response_events.request_id = projection_pending_approvals.request_id ) WHERE EXISTS ( SELECT 1 FROM latest_response_events WHERE latest_response_events.request_id = projection_pending_approvals.request_id ) `; yield* sql` WITH latest_stale_failures AS ( SELECT failure.request_id, failure.resolved_at FROM ( SELECT json_extract(payload_json, '$.requestId') AS request_id, created_at AS resolved_at, ROW_NUMBER() OVER ( PARTITION BY json_extract(payload_json, '$.requestId') ORDER BY created_at DESC, activity_id DESC ) AS row_number FROM projection_thread_activities WHERE kind = 'provider.approval.respond.failed' AND json_extract(payload_json, '$.requestId') IS NOT NULL AND ( lower(COALESCE(json_extract(payload_json, '$.detail'), '')) LIKE '%stale pending approval request%' OR lower(COALESCE(json_extract(payload_json, '$.detail'), '')) LIKE '%unknown pending approval request%' OR lower(COALESCE(json_extract(payload_json, '$.detail'), '')) LIKE '%unknown pending permission request%' ) ) AS failure WHERE failure.row_number = 1 ) UPDATE projection_pending_approvals SET status = 'resolved', decision = NULL, resolved_at = ( SELECT latest_stale_failures.resolved_at FROM latest_stale_failures WHERE latest_stale_failures.request_id = projection_pending_approvals.request_id ) WHERE status = 'pending' AND EXISTS ( SELECT 1 FROM latest_stale_failures WHERE latest_stale_failures.request_id = projection_pending_approvals.request_id ) `; yield* sql` UPDATE projection_threads SET latest_user_message_at = ( SELECT MAX(message.created_at) FROM projection_thread_messages AS message WHERE message.thread_id = projection_threads.thread_id AND message.role = 'user' ), pending_approval_count = COALESCE(( SELECT COUNT(*) FROM projection_pending_approvals WHERE projection_pending_approvals.thread_id = projection_threads.thread_id AND projection_pending_approvals.status = 'pending' ), 0), pending_user_input_count = COALESCE(( WITH latest_user_input_states AS ( SELECT latest.request_id, latest.kind, latest.detail FROM ( SELECT json_extract(activity.payload_json, '$.requestId') AS request_id, activity.kind, lower(COALESCE(json_extract(activity.payload_json, '$.detail'), '')) AS detail, ROW_NUMBER() OVER ( PARTITION BY json_extract(activity.payload_json, '$.requestId') ORDER BY activity.created_at DESC, activity.activity_id DESC ) AS row_number FROM projection_thread_activities AS activity WHERE activity.thread_id = projection_threads.thread_id AND json_extract(activity.payload_json, '$.requestId') IS NOT NULL AND activity.kind IN ( 'user-input.requested', 'user-input.resolved', 'provider.user-input.respond.failed' ) ) AS latest WHERE latest.row_number = 1 ) SELECT COUNT(*) FROM latest_user_input_states WHERE latest_user_input_states.kind = 'user-input.requested' OR ( latest_user_input_states.kind = 'provider.user-input.respond.failed' AND latest_user_input_states.detail NOT LIKE '%stale pending user-input request%' AND latest_user_input_states.detail NOT LIKE '%unknown pending user-input request%' ) ), 0), has_actionable_proposed_plan = COALESCE(( SELECT CASE WHEN projection_threads.latest_turn_id IS NOT NULL AND EXISTS ( SELECT 1 FROM projection_thread_proposed_plans AS latest_turn_plan_exists WHERE latest_turn_plan_exists.thread_id = projection_threads.thread_id AND latest_turn_plan_exists.turn_id = projection_threads.latest_turn_id ) THEN CASE WHEN ( SELECT latest_turn_plan.implemented_at FROM projection_thread_proposed_plans AS latest_turn_plan WHERE latest_turn_plan.thread_id = projection_threads.thread_id AND latest_turn_plan.turn_id = projection_threads.latest_turn_id ORDER BY latest_turn_plan.updated_at DESC, latest_turn_plan.plan_id DESC LIMIT 1 ) IS NULL THEN 1 ELSE 0 END WHEN EXISTS ( SELECT 1 FROM projection_thread_proposed_plans AS any_plan WHERE any_plan.thread_id = projection_threads.thread_id ) THEN CASE WHEN ( SELECT latest_plan.implemented_at FROM projection_thread_proposed_plans AS latest_plan WHERE latest_plan.thread_id = projection_threads.thread_id ORDER BY latest_plan.updated_at DESC, latest_plan.plan_id DESC LIMIT 1 ) IS NULL THEN 1 ELSE 0 END ELSE 0 END ), 0) `; });