feat(ql3): expose bounded worker session observation

2026-09-22 02:27:44 +08:00 · 2026-08-20 12:55:44 +08:00
parent 6265e31dce
commit 4a4fa85f51
24 changed files with 1437 additions and 28 deletions
@@ -348,5 +348,10 @@ export const postgresqlMainMigrationManifest: MigrationStreamManifest =
        checksum:
          '2fcac38386581189db63faacff325356f11c4529a8db9cef6be1a1ca706aaf10',
      }),
+      Object.freeze({
+        id: 'pg-0069-worker-session-management-observation',
+        checksum:
+          '1191255575589abc2686b391827607abddb4edb78007245dbaaf45dc1c4e5e8b',
+      }),
    ]),
  });
@@ -71,6 +71,7 @@ import { pg0065ApprovedActionManualRecoveryMigration } from '../approved-action/
 import { pg0066CancellationDispatchMigration } from '../run/migrations/pg-0066-cancellation-dispatch';
 import { pg0067CancellationDispatchManagementMigration } from '../run-management/pg-0067-cancellation-dispatch-management';
 import { pg0068CancellationDispatchProjectKeysetMigration } from '../run-management/pg-0068-cancellation-dispatch-project-keyset';
+import { pg0069WorkerSessionManagementObservationMigration } from '../remote-execution/pg-0069-worker-session-management-observation';

 export const postgresqlMainMigrationStream: MigrationStreamDefinition<PostgresMigrationContext> =
  Object.freeze({
@@ -147,5 +148,6 @@ export const postgresqlMainMigrationStream: MigrationStreamDefinition<PostgresMi
      pg0066CancellationDispatchMigration,
      pg0067CancellationDispatchManagementMigration,
      pg0068CancellationDispatchProjectKeysetMigration,
+      pg0069WorkerSessionManagementObservationMigration,
    ]),
  });
@@ -0,0 +1,17 @@
+import { CAPABILITIES_V67 } from '../run-management/pg-0068-cancellation-dispatch-project-keyset';
+import { definePostgresSqlMigration } from '../migrations/sqlMigration';
+
+export const CAPABILITIES_V68 = CAPABILITIES_V67.replace(
+  '"worker_session":1}',
+  '"worker_session":1,"worker_session_observation":1}',
+);
+
+export const pg0069WorkerSessionManagementObservationMigration =
+  definePostgresSqlMigration({
+    id: 'pg-0069-worker-session-management-observation',
+    statements: [
+      `REVOKE ALL ON "ql3"."worker_sessions" FROM ql3_worker_credential_manager`,
+      `GRANT SELECT ON "ql3"."worker_sessions" TO ql3_worker_credential_manager`,
+      `DO $ql3$ BEGIN UPDATE "ql3"."schema_capabilities" SET contract_version = 68, migration_id = 'pg-0069-worker-session-management-observation', capabilities = '${CAPABILITIES_V68}'::jsonb, updated_at_ms = floor(extract(epoch FROM transaction_timestamp()) * 1000)::bigint WHERE contract_name = 'control-core' AND contract_version = 67 AND migration_id = 'pg-0068-cancellation-dispatch-project-keyset' AND capabilities = '${CAPABILITIES_V67}'::jsonb; IF NOT FOUND THEN RAISE EXCEPTION 'control-core capability is not at version 67' USING ERRCODE = 'check_violation'; END IF; END $ql3$`,
+    ],
+  });
@@ -0,0 +1,188 @@
+import type { PostgresPool } from '@qinglong/runtime-core';
+import {
+  WORKER_SESSION_STATUSES,
+  assertWorkerCapabilitiesSnapshot,
+  assertWorkerConcurrency,
+  assertWorkerId,
+  assertWorkerSessionId,
+  type WorkerSessionRecord,
+  type WorkerSessionStatus,
+} from '@qinglong/runtime-core/worker-session';
+import {
+  MAX_WORKER_SESSION_OBSERVATION_PAGE_SIZE,
+  projectWorkerSessionObservation,
+  summarizeWorkerSessionObservation,
+  type WorkerSessionInspection,
+  type WorkerSessionObservationPage,
+} from '@qinglong/runtime-core/worker-session-observation';
+
+type Row = Record<string, unknown>;
+
+const SELECT_COLUMNS = `
+  worker_id AS "workerId",
+  session_id AS "sessionId",
+  generation AS "generation",
+  status AS "status",
+  version AS "version",
+  capabilities_json AS "capabilitiesJson",
+  capabilities_hash AS "capabilitiesHash",
+  max_concurrent_runs AS "maxConcurrentRuns",
+  available_slots AS "availableSlots",
+  registered_at_ms AS "registeredAtMs",
+  last_heartbeat_at_ms AS "lastHeartbeatAtMs",
+  lease_expires_at_ms AS "leaseExpiresAtMs",
+  updated_at_ms AS "updatedAtMs"
+`.trim();
+
+function string(row: Row, key: string): string {
+  const value = row[key];
+  if (typeof value !== 'string' || value.length === 0) {
+    throw new TypeError(`PostgreSQL Worker observation ${key} is invalid`);
+  }
+  return value;
+}
+
+function integer(row: Row, key: string): number {
+  const value = row[key];
+  const normalized =
+    typeof value === 'string' && /^(0|[1-9]\d*)$/.test(value)
+      ? Number(value)
+      : value;
+  if (!Number.isSafeInteger(normalized) || (normalized as number) < 0) {
+    throw new TypeError(`PostgreSQL Worker observation ${key} is invalid`);
+  }
+  return normalized as number;
+}
+
+function record(row: Row): Readonly<WorkerSessionRecord> {
+  const status = string(row, 'status');
+  if (!WORKER_SESSION_STATUSES.includes(status as WorkerSessionStatus)) {
+    throw new TypeError('PostgreSQL Worker observation status is invalid');
+  }
+  const worker: WorkerSessionRecord = Object.freeze({
+    workerId: string(row, 'workerId'),
+    sessionId: string(row, 'sessionId'),
+    generation: integer(row, 'generation'),
+    status: status as WorkerSessionStatus,
+    version: integer(row, 'version'),
+    capabilitiesJson: string(row, 'capabilitiesJson'),
+    capabilitiesHash: string(row, 'capabilitiesHash'),
+    maxConcurrentRuns: integer(row, 'maxConcurrentRuns'),
+    availableSlots: integer(row, 'availableSlots'),
+    registeredAtMs: integer(row, 'registeredAtMs'),
+    lastHeartbeatAtMs: integer(row, 'lastHeartbeatAtMs'),
+    leaseExpiresAtMs: integer(row, 'leaseExpiresAtMs'),
+    updatedAtMs: integer(row, 'updatedAtMs'),
+  });
+  assertWorkerId(worker.workerId);
+  assertWorkerSessionId(worker.sessionId);
+  assertWorkerCapabilitiesSnapshot(
+    worker.capabilitiesJson,
+    worker.capabilitiesHash,
+  );
+  assertWorkerConcurrency(worker.maxConcurrentRuns, worker.availableSlots);
+  return worker;
+}
+
+function observedAtMs(row: Row): number {
+  return integer(row, 'observedAtMs');
+}
+
+export class PostgresWorkerSessionObservationRepository {
+  constructor(private readonly pool: PostgresPool) {
+    if (!pool || typeof pool.query !== 'function') {
+      throw new TypeError('PostgreSQL Worker observation Pool is invalid');
+    }
+  }
+
+  async inspect(workerId: string): Promise<Readonly<WorkerSessionInspection>> {
+    assertWorkerId(workerId);
+    const result = await this.pool.query<Row>(
+      `WITH observation AS (
+         SELECT floor(extract(epoch FROM statement_timestamp()) * 1000)::bigint AS observed_at_ms
+       )
+       SELECT observation.observed_at_ms AS "observedAtMs", worker.*
+       FROM observation
+       LEFT JOIN LATERAL (
+         SELECT ${SELECT_COLUMNS}
+         FROM "ql3"."worker_sessions"
+         WHERE worker_id = $1
+         LIMIT 2
+       ) AS worker ON TRUE`,
+      [workerId],
+    );
+    if (result.rows.length !== 1) {
+      throw new TypeError('PostgreSQL Worker inspection violated its bound');
+    }
+    const observed = observedAtMs(result.rows[0]!);
+    if (
+      result.rows[0]!.workerId !== null &&
+      string(result.rows[0]!, 'workerId') !== workerId
+    ) {
+      throw new TypeError('PostgreSQL Worker inspection identity drifted');
+    }
+    return Object.freeze({
+      observedAtMs: observed,
+      worker:
+        result.rows[0]!.workerId === null
+          ? null
+          : projectWorkerSessionObservation(record(result.rows[0]!), observed),
+    });
+  }
+
+  async list(
+    afterWorkerId: string | null,
+  ): Promise<Readonly<WorkerSessionObservationPage>> {
+    if (afterWorkerId !== null) assertWorkerId(afterWorkerId);
+    const result = await this.pool.query<Row>(
+      `WITH observation AS (
+         SELECT floor(extract(epoch FROM statement_timestamp()) * 1000)::bigint AS observed_at_ms
+       ), page AS (
+         SELECT ${SELECT_COLUMNS}
+         FROM "ql3"."worker_sessions"
+         WHERE ($1::varchar IS NULL OR worker_id > $1)
+         ORDER BY worker_id
+         LIMIT $2
+       )
+       SELECT observation.observed_at_ms AS "observedAtMs", page.*
+       FROM observation LEFT JOIN page ON TRUE
+       ORDER BY page."workerId"`,
+      [afterWorkerId, MAX_WORKER_SESSION_OBSERVATION_PAGE_SIZE + 1],
+    );
+    if (
+      result.rows.length < 1 ||
+      result.rows.length > MAX_WORKER_SESSION_OBSERVATION_PAGE_SIZE + 1
+    ) {
+      throw new TypeError('PostgreSQL Worker observation page violated its bound');
+    }
+    const observed = observedAtMs(result.rows[0]!);
+    if (result.rows.some((row) => observedAtMs(row) !== observed)) {
+      throw new TypeError('PostgreSQL Worker observation clock drifted');
+    }
+    const observations =
+      result.rows[0]!.workerId === null
+        ? []
+        : result.rows.map((row) =>
+            projectWorkerSessionObservation(record(row), observed),
+          );
+    for (let index = 0; index < observations.length; index += 1) {
+      const workerId = observations[index]!.workerId;
+      const previous = observations[index - 1]?.workerId ?? afterWorkerId;
+      if (previous !== null && workerId <= previous) {
+        throw new TypeError('PostgreSQL Worker observation page is unordered');
+      }
+    }
+    const page = observations.slice(
+      0,
+      MAX_WORKER_SESSION_OBSERVATION_PAGE_SIZE,
+    );
+    return Object.freeze({
+      observedAtMs: observed,
+      workers: Object.freeze(page.map(summarizeWorkerSessionObservation)),
+      nextCursor:
+        observations.length > MAX_WORKER_SESSION_OBSERVATION_PAGE_SIZE
+          ? page.at(-1)!.workerId
+          : null,
+    });
+  }
+}
@@ -21,8 +21,8 @@ export interface PostgresSchemaContractTrigger {
 export interface PostgresSchemaContract {
  readonly schema: 'ql3';
  readonly contractName: 'control-core';
-  readonly contractVersion: 67;
-  readonly migrationId: 'pg-0068-cancellation-dispatch-project-keyset';
+  readonly contractVersion: 68;
+  readonly migrationId: 'pg-0069-worker-session-management-observation';
  readonly minimumServerMajor: 16;
  readonly maximumServerMajor: 18;
  readonly capabilities: Readonly<{
@@ -92,6 +92,7 @@ export interface PostgresSchemaContract {
    tool_invocation_artifact: 1;
    trigger_definition: 1;
    worker_session: 1;
+    worker_session_observation: 1;
    worker_credential: 1;
    worker_credential_delivery: 1;
    worker_credential_execution_receipt: 1;
@@ -121,8 +122,8 @@ export const postgresqlControlSchemaContract: PostgresSchemaContract =
  Object.freeze({
    schema: 'ql3',
    contractName: 'control-core',
-    contractVersion: 67,
-    migrationId: 'pg-0068-cancellation-dispatch-project-keyset',
+    contractVersion: 68,
+    migrationId: 'pg-0069-worker-session-management-observation',
    minimumServerMajor: 16,
    maximumServerMajor: 18,
    capabilities: Object.freeze({
@@ -192,6 +193,7 @@ export const postgresqlControlSchemaContract: PostgresSchemaContract =
      tool_invocation_artifact: 1,
      trigger_definition: 1,
      worker_session: 1,
+      worker_session_observation: 1,
      worker_credential: 1,
      worker_credential_delivery: 1,
      worker_credential_execution_receipt: 1,
@@ -1509,7 +1509,8 @@ const REQUIRED_WORKER_CREDENTIAL_MANAGER_PRIVILEGES: RequiredPrivileges =
          name === 'schema_migrations' ||
            name === 'schema_capabilities' ||
            name === 'projects' ||
-            name === 'project_role_bindings'
+            name === 'project_role_bindings' ||
+            name === 'worker_sessions'
            ? { ...NO_TABLE_PRIVILEGES, select: true }
            : name === 'worker_credential_management_plans' ||
              name === 'security_audit_events'
@@ -10,6 +10,7 @@ const OPERATIONS = [
  'worker-credential.propose',
  'worker-credential.decide',
  'worker-credential.inspect',
+  'worker-session.observe',
 ] as const;
 type Operation = (typeof OPERATIONS)[number];
 type Row = Record<string, unknown>;
@@ -31,6 +32,7 @@ const DEFAULT_LIMITS: Readonly<Record<Operation, number>> = Object.freeze({
  'worker-credential.propose': 30,
  'worker-credential.decide': 60,
  'worker-credential.inspect': 600,
+  'worker-session.observe': 600,
 });
 const ID = /^[A-Za-z0-9][A-Za-z0-9._:/-]{0,254}$/;
 const PROJECT = /^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$/;
@@ -44,6 +44,7 @@ export { postgresqlMainMigrationManifest } from '../migration/migrationManifest'
 export { PostgresApprovalRequestRepository } from '../approved-action/approvalRequestRepository';
 export { PostgresProjectPolicyRepository } from '../security/projectPolicyRepository';
 export { PostgresWorkerCredentialManagementPlanRepository } from './workerCredentialManagementPlanRepository';
+export { PostgresWorkerSessionObservationRepository } from '../remote-execution/workerSessionObservationRepository';
 export {
  PostgresPluginPackageIdentityKeysetLedgerRepository as PostgresWorkerCredentialManagementIdentityKeysetLedgerRepository,
  type ClusterManagementIdentityAuthority,
@@ -119,6 +119,7 @@ test('defines the immutable PostgreSQL capability and Run core stream', async ()
      'pg-0066-cancellation-dispatch',
      'pg-0067-cancellation-dispatch-management',
      'pg-0068-cancellation-dispatch-project-keyset',
+      'pg-0069-worker-session-management-observation',
    ],
  );
  for (const migration of postgresqlMainMigrationStream.migrations) {
@@ -597,6 +598,11 @@ test('freezes every published PostgreSQL migration checksum', () => {
      checksum:
        '2fcac38386581189db63faacff325356f11c4529a8db9cef6be1a1ca706aaf10',
    },
+    {
+      id: 'pg-0069-worker-session-management-observation',
+      checksum:
+        '1191255575589abc2686b391827607abddb4edb78007245dbaaf45dc1c4e5e8b',
+    },
  ];
  assert.deepEqual(
    postgresqlMainMigrationStream.migrations.map(({ id, checksum }) => ({
@@ -2421,3 +2427,36 @@ test('advances capability v67 with a Project-scoped blocked keyset', async () =>
    /migration_id = 'pg-0067-cancellation-dispatch-management'/,
  );
 });
+
+test('advances capability v68 with read-only Worker session observation', async () => {
+  const migration = migrationById(
+    'pg-0069-worker-session-management-observation',
+  );
+  const statements = [];
+  await migration.up({
+    async query(statement) {
+      statements.push(statement);
+      return { rows: [] };
+    },
+  });
+  const sql = statements.join('\n');
+  assert.match(
+    sql,
+    /REVOKE ALL ON "ql3"\."worker_sessions" FROM ql3_worker_credential_manager/,
+  );
+  assert.match(
+    sql,
+    /GRANT SELECT ON "ql3"\."worker_sessions" TO ql3_worker_credential_manager/,
+  );
+  assert.doesNotMatch(
+    sql,
+    /GRANT (?:INSERT|UPDATE|DELETE|TRUNCATE)[^;]+worker_sessions[^;]+ql3_worker_credential_manager/,
+  );
+  assert.match(sql, /contract_version = 68/);
+  assert.match(sql, /"worker_session_observation":1/);
+  assert.match(sql, /contract_version = 67/);
+  assert.match(
+    sql,
+    /migration_id = 'pg-0068-cancellation-dispatch-project-keyset'/,
+  );
+});
@@ -561,6 +561,7 @@ function workerCredentialPrivileges(kind) {
      ? [
          'worker_credential_management_quota_buckets',
          'plugin_package_identity_keyset_ledger',
+          'worker_sessions',
        ]
      : []),
    ...(manager
@@ -835,7 +836,7 @@ test('accepts the exact PostgreSQL control schema and least-privilege runtime ro
    serverMajor: 16,
    currentUser: 'ql3_runtime',
    contractName: 'control-core',
-    contractVersion: 67,
+    contractVersion: 68,
    migrationIds: [
      'pg-0001-schema-capability',
      'pg-0002-run-core',
@@ -905,6 +906,7 @@ test('accepts the exact PostgreSQL control schema and least-privilege runtime ro
      'pg-0066-cancellation-dispatch',
      'pg-0067-cancellation-dispatch-management',
      'pg-0068-cancellation-dispatch-project-keyset',
+      'pg-0069-worker-session-management-observation',
    ],
  });
 });
@@ -935,10 +937,10 @@ test('accepts the exact schema and isolated least-privilege admin role', async (
    }),
  );
  assert.equal(report.currentUser, 'ql3_admin');
-  assert.equal(report.contractVersion, 67);
+  assert.equal(report.contractVersion, 68);
  assert.equal(
    report.migrationIds.at(-1),
-    'pg-0068-cancellation-dispatch-project-keyset',
+    'pg-0069-worker-session-management-observation',
  );
 });

@@ -951,10 +953,10 @@ test('accepts the isolated least-privilege automation manager role', async () =>
    }),
  );
  assert.equal(report.currentUser, 'ql3_automation_manager');
-  assert.equal(report.contractVersion, 67);
+  assert.equal(report.contractVersion, 68);
  assert.equal(
    report.migrationIds.at(-1),
-    'pg-0068-cancellation-dispatch-project-keyset',
+    'pg-0069-worker-session-management-observation',
  );

  const widened = automationManagerPrivileges();
@@ -983,10 +985,10 @@ test('accepts the isolated least-privilege human Approval manager role', async (
    }),
  );
  assert.equal(report.currentUser, 'ql3_approval_manager');
-  assert.equal(report.contractVersion, 67);
+  assert.equal(report.contractVersion, 68);
  assert.equal(
    report.migrationIds.at(-1),
-    'pg-0068-cancellation-dispatch-project-keyset',
+    'pg-0069-worker-session-management-observation',
  );

  const widened = approvalManagerPrivileges();
@@ -1017,10 +1019,10 @@ test('accepts the isolated least-privilege Run manager role', async () => {
    }),
  );
  assert.equal(report.currentUser, 'ql3_run_manager');
-  assert.equal(report.contractVersion, 67);
+  assert.equal(report.contractVersion, 68);
  assert.equal(
    report.migrationIds.at(-1),
-    'pg-0068-cancellation-dispatch-project-keyset',
+    'pg-0069-worker-session-management-observation',
  );

  const widened = runManagerPrivileges();
@@ -1181,10 +1183,10 @@ test('accepts the exact schema and isolated Worker ingress role', async () => {
    }),
  );
  assert.equal(report.currentUser, 'ql3_worker_ingress');
-  assert.equal(report.contractVersion, 67);
+  assert.equal(report.contractVersion, 68);
  assert.equal(
    report.migrationIds.at(-1),
-    'pg-0068-cancellation-dispatch-project-keyset',
+    'pg-0069-worker-session-management-observation',
  );
 });

@@ -0,0 +1,119 @@
+const assert = require('node:assert/strict');
+const { test } = require('node:test');
+
+const {
+  canonicalRemoteWorkerCapabilities,
+} = require('@qinglong/runtime-core/remote-dispatch');
+const {
+  PostgresWorkerSessionObservationRepository,
+} = require('@qinglong/cluster-postgres/worker-credential-manager');
+
+function row(index, overrides = {}) {
+  const canonical = canonicalRemoteWorkerCapabilities(
+    overrides.capabilities ?? {
+      architecture: index % 2 === 0 ? 'arm64' : 's390x',
+      executors: ['remote-worker'],
+      protocolVersion: '1.0.0',
+      supportTier: index % 2 === 0 ? 'tier1' : 'candidate',
+      runtimes: [{ name: 'node', version: '24.18.0' }],
+    },
+  );
+  return {
+    observedAtMs: 2_000,
+    workerId: `worker-${String(index).padStart(2, '0')}`,
+    sessionId: `018f0f5d-7b6a-7a11-8f4d-${String(index + 1).padStart(
+      12,
+      '0',
+    )}`,
+    generation: 1,
+    status: 'online',
+    version: 2,
+    capabilitiesJson: canonical.json,
+    capabilitiesHash: canonical.hash,
+    maxConcurrentRuns: 4,
+    availableSlots: 2,
+    registeredAtMs: 1_000,
+    lastHeartbeatAtMs: 1_900,
+    leaseExpiresAtMs: 3_000,
+    updatedAtMs: 1_900,
+    ...Object.fromEntries(
+      Object.entries(overrides).filter(([key]) => key !== 'capabilities'),
+    ),
+  };
+}
+
+test('inspects one exact Worker without exposing raw capability content', async () => {
+  const queries = [];
+  const repository = new PostgresWorkerSessionObservationRepository({
+    async query(text, values) {
+      queries.push({ text, values });
+      return { rows: [row(0)], rowCount: 1 };
+    },
+  });
+  const result = await repository.inspect('worker-00');
+  assert.equal(result.observedAtMs, 2_000);
+  assert.equal(result.worker.workerId, 'worker-00');
+  assert.equal(result.worker.compatibility, 'default_placement');
+  assert.deepEqual(result.worker.runtimes, [
+    { name: 'node', version: '24.18.0' },
+  ]);
+  assert.equal(Object.hasOwn(result.worker, 'capabilitiesJson'), false);
+  assert.deepEqual(queries[0].values, ['worker-00']);
+  assert.match(queries[0].text, /LIMIT 2/);
+});
+
+test('returns a masked absent inspection and a fixed sixteen-item keyset page', async () => {
+  let call = 0;
+  const repository = new PostgresWorkerSessionObservationRepository({
+    async query(_text, values) {
+      call += 1;
+      if (call === 1) {
+        return {
+          rows: [{ observedAtMs: 2_000, workerId: null }],
+          rowCount: 1,
+        };
+      }
+      assert.deepEqual(values, [null, 17]);
+      return {
+        rows: Array.from({ length: 17 }, (_, index) => row(index)),
+        rowCount: 17,
+      };
+    },
+  });
+  assert.deepEqual(await repository.inspect('missing-worker'), {
+    observedAtMs: 2_000,
+    worker: null,
+  });
+  const page = await repository.list(null);
+  assert.equal(page.workers.length, 16);
+  assert.equal(page.nextCursor, 'worker-15');
+  assert.equal(page.workers[1].compatibility, 'explicit_placement_required');
+  assert.equal(Object.hasOwn(page.workers[0], 'runtimes'), false);
+  assert.equal(Object.hasOwn(page.workers[0], 'declaredCapacity'), false);
+});
+
+test('fails closed on identity, ordering and database-clock drift', async () => {
+  const identity = new PostgresWorkerSessionObservationRepository({
+    async query() {
+      return { rows: [row(1)], rowCount: 1 };
+    },
+  });
+  await assert.rejects(identity.inspect('worker-00'), /identity drifted/);
+
+  const unordered = new PostgresWorkerSessionObservationRepository({
+    async query() {
+      return { rows: [row(1), row(0)], rowCount: 2 };
+    },
+  });
+  await assert.rejects(unordered.list(null), /page is unordered/);
+
+  const clock = new PostgresWorkerSessionObservationRepository({
+    async query() {
+      return {
+        rows: [row(0), row(1, { observedAtMs: 2_001 })],
+        rowCount: 2,
+      };
+    },
+  });
+  await assert.rejects(clock.list(null), /clock drifted/);
+});