並列エージェントに実行環境をリースする — ポート帯・DBスキーマ・devサーバを1エージェント1環境で分離する設計

// lease.mjs — エージェント1体分の実行環境をリースする
import { writeFileSync } from 'node:fs';
import pg from 'pg';
 
// 1エージェントが使うポート数。dev(3000系)/api(4000系)/db-proxy(5000系)を
// まとめて1帯として確保する。10あれば周辺ツールの分も足りる。
const PORTS_PER_AGENT = 10;
const BASE = { dev: 3000, api: 4000, dbProxy: 5000 };
 
// エージェント番号(0,1,2,...)から決定的にポート帯を導く。
// 番号さえ固定すれば、再起動しても毎回同じポートに戻る点が重要。
function derivePorts(agentIndex) {
  const offset = agentIndex * PORTS_PER_AGENT;
  return {
    dev: BASE.dev + offset,       // 例: agent0=3000, agent1=3010
    api: BASE.api + offset,
    dbProxy: BASE.dbProxy + offset,
  };
}
 
// 専用スキーマを「移行済みテンプレート」から複製する。
// 毎回 migrate を走らせない。ここが並列化の最大のボトルネックだった。
async function provisionSchema(client, schema, template = 'tpl_migrated') {
  // DROP SCHEMA は識別子をプレースホルダにできないため、連結の前に文字種を絞る
  for (const name of [schema, template]) {
    if (!/^[a-z0-9_]+$/.test(name)) {
      throw new Error(`unsafe schema name: ${name}`);
    }
  }
  // 既存が残っていれば掃除してから、テンプレートの構造を複製する
  await client.query(`DROP SCHEMA IF EXISTS ${schema} CASCADE`);
  // clone_schema の引数は識別子ではなく値なので、プレースホルダで渡せる
  await client.query('SELECT clone_schema($1, $2)', [template, schema]);
  return schema;
}
 
export async function acquireLease(agentIndex, agentId) {
  const ports = derivePorts(agentIndex);
  const schema = `agent_${agentIndex}`;
 
  const client = new pg.Client({ connectionString: process.env.ADMIN_DATABASE_URL });
  await client.connect();
  try {
    await provisionSchema(client, schema);
  } finally {
    await client.end();
  }
 
  // search_path でスキーマを差し込む。アプリ側はテーブル名を変えなくてよい。
  const dbUrl =
    `${process.env.APP_DATABASE_URL}?options=-c%20search_path%3D${schema}`;
 
  const env = [
    `PORT=${ports.dev}`,
    `API_PORT=${ports.api}`,
    `DB_PROXY_PORT=${ports.dbProxy}`,
    `DATABASE_URL=${dbUrl}`,
    `AGENT_ID=${agentId}`,
    `LEASE_INDEX=${agentIndex}`,
  ].join('\n');
 
  writeFileSync(`.env.agent.${agentIndex}`, env + '\n');
  return { agentIndex, agentId, schema, ports, envFile: `.env.agent.${agentIndex}` };
}

-- 移行済みテンプレートの構造を、別スキーマに複製する関数。
-- search_path を pg_catalog に固定しているのは、既定値や制約の定義が
-- 必ずスキーマ修飾つきで描画されるようにするためです（後段の貼り替えの前提）。
CREATE OR REPLACE FUNCTION clone_schema(src text, dst text)
RETURNS void AS $$
DECLARE
  obj record;
BEGIN
  EXECUTE format('CREATE SCHEMA %I', dst);
 
  -- 1) シーケンスを先に用意する。serial 列の既定値をあとでここへ向け直す
  FOR obj IN
    SELECT sequencename FROM pg_sequences WHERE schemaname = src
  LOOP
    EXECUTE format('CREATE SEQUENCE %I.%I', dst, obj.sequencename);
  END LOOP;
 
  -- 2) テーブル構造を複製する。主キー・一意制約・CHECK・索引・既定値はここで入る
  FOR obj IN
    SELECT tablename FROM pg_tables WHERE schemaname = src
  LOOP
    EXECUTE format(
      'CREATE TABLE %I.%I (LIKE %I.%I INCLUDING ALL)',
      dst, obj.tablename, src, obj.tablename
    );
  END LOOP;
 
  -- 3) serial 列の既定値は src のシーケンスを指したまま複製されるので、
  --    nextval の宛先を dst 側へ貼り替える
  FOR obj IN
    SELECT c.relname AS tbl, a.attname AS col,
           pg_get_expr(d.adbin, d.adrelid) AS def
    FROM pg_class c
    JOIN pg_namespace n ON n.oid = c.relnamespace
    JOIN pg_attribute a ON a.attrelid = c.oid AND a.attnum > 0
    JOIN pg_attrdef  d ON d.adrelid = c.oid AND d.adnum = a.attnum
    WHERE n.nspname = dst
      AND c.relkind = 'r'
      AND pg_get_expr(d.adbin, d.adrelid) LIKE 'nextval(%'
  LOOP
    EXECUTE format(
      'ALTER TABLE %I.%I ALTER COLUMN %I SET DEFAULT %s',
      dst, obj.tbl, obj.col,
      replace(obj.def, quote_ident(src) || '.', quote_ident(dst) || '.')
    );
  END LOOP;
 
  -- 4) 外部キーは LIKE では複製されない。src の定義から張り直す
  FOR obj IN
    SELECT c.relname AS tbl, con.conname AS name,
           pg_get_constraintdef(con.oid) AS def
    FROM pg_constraint con
    JOIN pg_class c     ON c.oid = con.conrelid
    JOIN pg_namespace n ON n.oid = c.relnamespace
    WHERE n.nspname = src AND con.contype = 'f'
  LOOP
    EXECUTE format(
      'ALTER TABLE %I.%I ADD CONSTRAINT %I %s',
      dst, obj.tbl, obj.name,
      replace(obj.def, quote_ident(src) || '.', quote_ident(dst) || '.')
    );
  END LOOP;
END;
$$ LANGUAGE plpgsql SET search_path = pg_catalog;

-- テンプレートと複製先で、制約の本数が一致するかを確かめる
SELECT n.nspname AS schema,
       count(*) FILTER (WHERE con.contype = 'f') AS foreign_keys,
       count(*) FILTER (WHERE con.contype = 'c') AS check_constraints
FROM pg_constraint con
JOIN pg_class c     ON c.oid = con.conrelid
JOIN pg_namespace n ON n.oid = c.relnamespace
WHERE n.nspname IN ('tpl_migrated', 'agent_0')
GROUP BY n.nspname
ORDER BY n.nspname;

# エージェント2の帯(3020)に残ったプロセスを確実に掃除する。
# 決定的ポートだからこそ、掃除の対象が毎回同じで迷わない。
lsof -ti:3020 | xargs -r kill -9

// reap.mjs — 期限切れのリースを回収する（定期実行 or 起動時に走らせる）
import pg from 'pg';
import { execSync } from 'node:child_process';
 
const PORTS_PER_AGENT = 10;
const DEV_BASE = 3000;
 
export async function reapLease(agentIndex) {
  // 1) スキーマを落とす
  const client = new pg.Client({ connectionString: process.env.ADMIN_DATABASE_URL });
  await client.connect();
  try {
    await client.query(`DROP SCHEMA IF EXISTS agent_${agentIndex} CASCADE`);
  } finally {
    await client.end();
  }
 
  // 2) 帯に残ったプロセスを掃除する（決定的ポートなので対象が確定する）
  const devPort = DEV_BASE + agentIndex * PORTS_PER_AGENT;
  try {
    execSync(`lsof -ti:${devPort} | xargs -r kill -9`, { stdio: 'ignore' });
  } catch {
    // 掃除対象がなければ何もしない。ここで失敗しても回収は続行する
  }
}

// lease.mjs に追記 — worktree もリースの一部として払い出す
import { execFileSync } from 'node:child_process';
import { existsSync } from 'node:fs';
 
// エージェント専用の worktree を用意する。
// ブランチ名も番号から決定的に導き、回収時に対象で迷わないようにする。
function provisionWorktree(agentIndex, baseRef = 'origin/main') {
  const dir = `../wt-agent-${agentIndex}`;
  const branch = `agent/${agentIndex}`;
 
  if (existsSync(dir)) return dir; // 再取得は冪等にしておく
 
  // -B は同名ブランチを作り直す。前回の残骸があっても baseRef の地点に戻る
  execFileSync('git', ['worktree', 'add', '-B', branch, dir, baseRef], {
    stdio: 'inherit',
  });
  return dir;
}

// reap.mjs に追記 — worktree を登録ごと外す
execFileSync('git', ['worktree', 'remove', '--force', `../wt-agent-${agentIndex}`], {
  stdio: 'ignore',
});
execFileSync('git', ['worktree', 'prune'], { stdio: 'ignore' });

#!/usr/bin/env bash
# ready.sh <agentIndex> — devサーバが応答してからタスクを渡す
set -euo pipefail
IDX="$1"; PORT=$((3000 + IDX * 10)); DEADLINE=$((SECONDS + 60))
 
until curl -sf "http://127.0.0.1:${PORT}/api/health" >/dev/null; do
  if (( SECONDS > DEADLINE )); then
    echo "agent ${IDX}: dev server not ready on ${PORT}" >&2
    exit 1
  fi
  sleep 1
done
echo "agent ${IDX}: ready on ${PORT}"