next-tjwater-server/resources/sql/014_operational_tasks.sql

-- Durable project-scoped operational tasks, immutable artifacts and approvals.

BEGIN;

CREATE SCHEMA IF NOT EXISTS workbench;

CREATE TABLE IF NOT EXISTS workbench.operational_tasks (
    project_id uuid NOT NULL,
    task_id uuid NOT NULL,
    create_request_id uuid NOT NULL,
    task_type text NOT NULL CHECK (task_type IN
        ('scheduled_flow','external_work_order','event_investigation')),
    workflow text NOT NULL CHECK (length(workflow) > 0),
    status text NOT NULL CHECK (status IN
        ('queued','running','waiting_human','completed','failed','cancelled')),
    priority text NOT NULL CHECK (priority IN ('normal','urgent')),
    revision bigint NOT NULL CHECK (revision >= 1),
    execution_generation bigint NOT NULL CHECK (execution_generation >= 1),
    execution_request_id uuid NOT NULL,
    agent_session_id text,
    agent_lease_expires_at timestamptz,
    current_plan_revision bigint,
    current_report_revision bigint,
    scheduled_for timestamptz,
    created_at timestamptz NOT NULL,
    updated_at timestamptz NOT NULL,
    document jsonb NOT NULL,
    PRIMARY KEY (project_id, task_id),
    UNIQUE (project_id, create_request_id),
    UNIQUE (project_id, execution_request_id),
    CHECK ((jsonb_typeof(document)='object'
        AND document->>'task_id'=task_id::text
        AND document->>'task_type'=task_type
        AND document->>'workflow'=workflow
        AND document->>'status'=status
        AND (document->>'revision')::bigint=revision) IS TRUE)
);

CREATE INDEX IF NOT EXISTS operational_tasks_status_idx
    ON workbench.operational_tasks(project_id,status,updated_at DESC,task_id);
CREATE INDEX IF NOT EXISTS operational_tasks_lease_idx
    ON workbench.operational_tasks(project_id,agent_lease_expires_at)
    WHERE status='running' AND agent_lease_expires_at IS NOT NULL;
CREATE UNIQUE INDEX IF NOT EXISTS operational_tasks_external_source_idx
    ON workbench.operational_tasks(
        project_id,
        (document#>>'{source,kind}'),
        (document#>>'{source,external_id}')
    ) WHERE document#>>'{source,external_id}' IS NOT NULL;

CREATE TABLE IF NOT EXISTS workbench.operational_task_requests (
    project_id uuid NOT NULL,
    scope text NOT NULL CHECK (length(scope) > 0),
    request_id uuid NOT NULL,
    request_digest text NOT NULL CHECK (length(request_digest) > 0),
    response jsonb NOT NULL,
    created_at timestamptz NOT NULL DEFAULT now(),
    PRIMARY KEY (project_id,scope,request_id)
);

CREATE TABLE IF NOT EXISTS workbench.operational_task_evidence (
    project_id uuid NOT NULL,
    task_id uuid NOT NULL,
    evidence_id text NOT NULL CHECK (length(evidence_id) > 0),
    document jsonb NOT NULL,
    created_at timestamptz NOT NULL,
    PRIMARY KEY (project_id,task_id,evidence_id),
    FOREIGN KEY (project_id,task_id)
        REFERENCES workbench.operational_tasks(project_id,task_id) ON DELETE CASCADE
);

CREATE TABLE IF NOT EXISTS workbench.operational_task_plans (
    project_id uuid NOT NULL,
    task_id uuid NOT NULL,
    revision bigint NOT NULL CHECK (revision >= 1),
    document jsonb NOT NULL,
    created_at timestamptz NOT NULL,
    PRIMARY KEY (project_id,task_id,revision),
    FOREIGN KEY (project_id,task_id)
        REFERENCES workbench.operational_tasks(project_id,task_id) ON DELETE CASCADE
);

CREATE TABLE IF NOT EXISTS workbench.operational_task_reports (
    project_id uuid NOT NULL,
    task_id uuid NOT NULL,
    revision bigint NOT NULL CHECK (revision >= 1),
    document jsonb NOT NULL,
    created_at timestamptz NOT NULL,
    PRIMARY KEY (project_id,task_id,revision),
    FOREIGN KEY (project_id,task_id)
        REFERENCES workbench.operational_tasks(project_id,task_id) ON DELETE CASCADE
);

CREATE TABLE IF NOT EXISTS workbench.operational_task_actions (
    project_id uuid NOT NULL,
    task_id uuid NOT NULL,
    action_id uuid NOT NULL,
    version bigint NOT NULL CHECK (version >= 1),
    approval_status text NOT NULL CHECK (approval_status IN
        ('pending','approved','rejected')),
    execution_status text NOT NULL CHECK (execution_status IN
        ('not_submitted','submitted','succeeded','failed','unsupported','unknown')),
    document jsonb NOT NULL,
    created_at timestamptz NOT NULL,
    updated_at timestamptz NOT NULL,
    PRIMARY KEY (project_id,task_id,action_id),
    FOREIGN KEY (project_id,task_id)
        REFERENCES workbench.operational_tasks(project_id,task_id) ON DELETE CASCADE,
    CHECK ((document->>'action_id'=action_id::text
        AND (document->>'version')::bigint=version
        AND document->>'approval_status'=approval_status
        AND document->>'execution_status'=execution_status) IS TRUE)
);

CREATE TABLE IF NOT EXISTS workbench.operational_task_action_revisions (
    project_id uuid NOT NULL,
    task_id uuid NOT NULL,
    action_id uuid NOT NULL,
    version bigint NOT NULL CHECK (version >= 1),
    document jsonb NOT NULL,
    recorded_at timestamptz NOT NULL,
    PRIMARY KEY (project_id,task_id,action_id,version),
    FOREIGN KEY (project_id,task_id,action_id)
        REFERENCES workbench.operational_task_actions(project_id,task_id,action_id)
        ON DELETE CASCADE,
    CHECK ((document->>'action_id'=action_id::text
        AND (document->>'version')::bigint=version) IS TRUE)
);

CREATE TABLE IF NOT EXISTS workbench.operational_task_external_feedback (
    project_id uuid NOT NULL,
    task_id uuid NOT NULL,
    source_event_id text NOT NULL CHECK (length(source_event_id) > 0),
    source_revision bigint NOT NULL CHECK (source_revision >= 1),
    document jsonb NOT NULL,
    accepted_at timestamptz NOT NULL,
    PRIMARY KEY (project_id,task_id,source_event_id),
    FOREIGN KEY (project_id,task_id)
        REFERENCES workbench.operational_tasks(project_id,task_id) ON DELETE CASCADE
);

CREATE TABLE IF NOT EXISTS workbench.operational_task_notifications (
    project_id uuid NOT NULL,
    notification_id uuid NOT NULL,
    task_id uuid NOT NULL,
    created_at timestamptz NOT NULL,
    document jsonb NOT NULL,
    PRIMARY KEY (project_id,notification_id),
    FOREIGN KEY (project_id,task_id)
        REFERENCES workbench.operational_tasks(project_id,task_id) ON DELETE CASCADE
);

CREATE INDEX IF NOT EXISTS operational_task_notifications_unread_idx
    ON workbench.operational_task_notifications(project_id,created_at DESC);

CREATE TABLE IF NOT EXISTS workbench.operational_task_notification_reads (
    project_id uuid NOT NULL,
    notification_id uuid NOT NULL,
    user_id uuid NOT NULL,
    version bigint NOT NULL CHECK (version >= 2),
    read_at timestamptz NOT NULL,
    PRIMARY KEY (project_id,notification_id,user_id),
    FOREIGN KEY (project_id,notification_id)
        REFERENCES workbench.operational_task_notifications(project_id,notification_id)
        ON DELETE CASCADE
);

CREATE TABLE IF NOT EXISTS workbench.operational_task_schedules (
    project_id uuid NOT NULL,
    schedule_id uuid NOT NULL,
    create_request_id uuid NOT NULL,
    schedule_key text NOT NULL CHECK (length(schedule_key) > 0),
    next_run_at timestamptz NOT NULL,
    interval_seconds integer NOT NULL CHECK (interval_seconds >= 60),
    enabled boolean NOT NULL,
    version bigint NOT NULL CHECK (version >= 1),
    document jsonb NOT NULL,
    created_at timestamptz NOT NULL,
    updated_at timestamptz NOT NULL,
    PRIMARY KEY (project_id,schedule_id),
    UNIQUE (project_id,create_request_id),
    UNIQUE (project_id,schedule_key),
    CHECK ((document->>'schedule_id'=schedule_id::text
        AND document->>'schedule_key'=schedule_key
        AND (document->>'version')::bigint=version) IS TRUE)
);

CREATE TABLE IF NOT EXISTS workbench.operational_schedule_occurrences (
    project_id uuid NOT NULL,
    schedule_id uuid NOT NULL,
    due_at timestamptz NOT NULL,
    task_id uuid NOT NULL,
    created_at timestamptz NOT NULL,
    PRIMARY KEY (project_id,schedule_id,due_at),
    UNIQUE (project_id,task_id),
    FOREIGN KEY (project_id,schedule_id)
        REFERENCES workbench.operational_task_schedules(project_id,schedule_id),
    FOREIGN KEY (project_id,task_id)
        REFERENCES workbench.operational_tasks(project_id,task_id)
);

COMMIT;