Capstone 8 — Legacy Migration Agent

Project Brief

Meridian Public Records runs the UCCUniform Commercial Code — the body of US law governing secured transactions. A UCC-1 financing statement is the public filing a lender makes to claim an interest in a borrower's assets. Filings live with each state's Secretary of State office. filing system for a consortium of eleven Secretary of State offices. The system has been on Oracle since 2003. The database is 6 TB, the licence renewal is $1.4M a year, and the two DBAs who wrote the PL/SQL retired in 2021.

Leadership has approved a move to PostgreSQL. The systems integrator quoted 14 months and $2.1M — most of it human hours spent reading DDL, hand-translating packages, and reconciling row counts one table at a time.

Migration is a reading problem before it is a writing problem. Someone has to look at NUMBER(12,0) and decide whether it becomes bigint or numeric(12,0). Someone has to notice the DATE trap above. Someone has to know that Oracle treats the empty string as NULL and PostgreSQL does not.

That reading work is exactly what an agent is good at. It is also exactly where an agent will produce a plausible, confident, wrong answer if you let it. So this capstone is as much about the guardrails and the validator as it is about the translation — and the hardest thing you will build is the subagent that refuses to do its job.

Prerequisites

This is the last and hardest capstone in the course. It assumes everything.

Environment: Python 3.10+, Docker Desktop with ~6 GB free disk and 6 GB RAM allocated, and an ANTHROPIC_API_KEY.

Domain Glossary

Two vocabularies collide in this project. Here is the minimum of each.

Oracle terms

Migration terms

Architecture

One coordinator, five specialists, two MCP servers wrapping two live databases, three guardrails that run before any tool executes, and one gate that only a person can open.

Two things in that diagram are worth pausing on, because they are the design decisions rather than the boxes.

The coordinator never touches a database. It has access to all three MCP servers, and its system prompt forbids using them directly. Every read and every write goes through a specialist. That looks like ceremony until you ask what happens when a migration goes wrong at 2 a.m.: with specialists, the audit log tells you which subagent, on which object, with which instructions. With a coordinator that calls tools itself, it tells you that something called pg_apply_ddl.

The guardrails sit between the agent and the tools, not inside them. A validation check written inside pg_apply_ddl protects pg_apply_ddl. A can_use_tool callback protects every tool that exists now and every tool anyone adds later. That difference is the entire reason the SDK has the hook.

Animation 1: Where the Two Type Systems Diverge

Most Oracle types have an obvious PostgreSQL counterpart. The interesting ones are where the counterpart is obvious and wrong.

Read the two red cards again. Both are cases where the name matches and the behaviour does not, and both fail silently:

• DATE → date loads every row successfully and discards the time. Your row counts match. Your checksums are computed over the truncated values, so they match too. The only check that catches it is comparing actual values, row by row.
• VARCHAR2(60 BYTE) → varchar(60) works perfectly until the first row containing a non-ASCII character, which may be years after cutover.

Animation 2: The Five-Phase Pipeline

Five phases, strictly ordered, each gated on the one before. Then a gate that is not a phase.

Why this order, specifically

The ordering is not arbitrary and it is not just dependency-following. Two of the choices are worth defending:

Data before code. You could convert the PL/SQL first — it does not depend on the rows. But converting code you cannot test is how you accumulate a pile of plausible functions nobody has run. Move the data first and every converted function has real rows to execute against the moment it exists.

Largest table first, within phase 3. Counter-intuitive: you would rather start with an easy win. But the failure modes in a bulk load — disk, memory, encoding, LOB handling — scale with size. Migrating UCC_DEBTOR (7,418 rows) before STATE_SOS_SOURCE (11 rows) means a capacity problem surfaces in minute two instead of minute forty, after you have already believed five successful loads.

Animation 3: Coordinator and Five Specialists

Five specialists rather than one generalist buys three specific things:

Different models for different work. schema-translator and plsql-converter run Sonnet, because reading DDL and converting a package are genuinely hard. data-migrator, appsql-rewriter and migration-validator run Haiku, because their work is high-volume and mechanical. On a schema with 400 objects that routing is most of the bill.

Restricted tools per role. appsql-rewriter has scan_app_sql and write_artifact. It has no database access at all. It cannot touch a database, however creatively it is prompted, because the capability was never granted. That is a stronger property than instructing it not to.

Isolated context. Each specialist sees one object and its own instructions. It does not see the coordinator's transcript or the other four specialists' work. Fewer tokens, and — more usefully — no opportunity to be confused by a decision made about a different table.

Animation 4: The Empty-String Trap

If you remember one thing from this capstone, this is it.

Now look at what that does to a query that has been running correctly since 2011:

-- app/filing_repository.py :: debtors_missing_address_line_2()
SELECT d.debtor_id, d.debtor_name, d.city, d.state_code
  FROM ucc_debtor d
 WHERE d.mailing_address_2 IS NULL
 ORDER BY d.debtor_name;

-- Oracle: 1,412 rows. Correct.
-- (Oracle collapsed '' to NULL on the way in, so IS NULL catches both.)

-- Same query. Same table name. Same column. No error.
SELECT d.debtor_id, d.debtor_name, d.city, d.state_code
  FROM ucc_debtor d
 WHERE d.mailing_address_2 IS NULL
 ORDER BY d.debtor_name;

-- PostgreSQL: 0 rows.
--
-- Not an error. Not a warning. Not a constraint violation.
-- The compliance report that has said "1,412 incomplete records"
-- every month for nine years now says zero, and the first person
-- to notice will assume the data got better.

Animation 5: Package → Schema, and the One That Cannot Convert

PostgreSQL has no packages. The standard translation is a schema of the same name containing one function per public routine — which has the pleasant property that pkg_risk_calc.score_debtor(x) keeps working in application code without a single edit.

That one converts cleanly. This one does not:

-- legacy-oracle/03_packages.sql
PROCEDURE log_audit (p_filing_id IN NUMBER,
                     p_action    IN VARCHAR2,
                     p_detail    IN VARCHAR2) IS
  PRAGMA AUTONOMOUS_TRANSACTION;    -- <-- there is no PostgreSQL equivalent
BEGIN
  INSERT INTO filing_audit (audit_id, filing_id, action, detail)
  VALUES (seq_audit_id.NEXTVAL, p_filing_id, p_action, p_detail);
  COMMIT;                            -- commits INDEPENDENTLY of the caller
END log_audit;

The pragma is the entire point of the procedure. It means the audit row commits even when the business transaction that called it rolls back. That is what makes it an audit log rather than a diary of things that happened to succeed.

PostgreSQL cannot do this in-process. The options are dblink (open a second connection to yourself — works, and it is a connection-pool problem waiting to happen), a background worker, or moving the audit write out of the transaction entirely at the application layer. All three are design decisions with cost and blast radius. None of them is a translation.

Animation 6: Three Guards That Run Before the Tool Does

Every tool call passes through can_use_tool first. The guards return PermissionResultDeny, so the dangerous call never executes — as opposed to a PostToolUse hook, which would give you an excellent record of the DROP TABLE that already happened.

Guard 1: allow-list, not deny-list

The obvious way to make Oracle read-only is to block the dangerous verbs. Write a regex for DROP|DELETE|UPDATE|INSERT|TRUNCATE and refuse anything that matches.

That approach loses. Not dramatically — it will catch the first several things you think of. It loses on MERGE, and FLASHBACK, and PURGE, and COMMENT ON, and LOCK TABLE, and a BEGIN ... END; block with an UPDATE buried three lines in, and whatever Oracle adds next release. Every deny-list is a list of the attacks you already imagined.

So the guard inverts it. Three shapes are permitted — statements starting with SELECT or WITH, DBMS_METADATA calls, and dictionary-view reads — and everything else is denied by construction. New Oracle feature, novel phrasing, creative prompt: all denied, because none of them are on the list.

# solution/hooks.py
from claude_agent_sdk import PermissionResultAllow, PermissionResultDeny

_ORACLE_READ_PREFIXES = ("select", "with")
_ORACLE_READ_CALLS = ("dbms_metadata.get_ddl", "dbms_metadata.set_transform_param")
_ORACLE_DICTIONARY = re.compile(r"\b(all|user|dba)_[a-z_]+\b", re.I)

async def enforce_oracle_readonly(tool_name: str, tool_input: dict, context):
    if not tool_name.startswith("mcp__oracle_src__"):
        return PermissionResultAllow()          # not our business

    statement = _statement_of(tool_input).strip()
    if not statement:
        return PermissionResultAllow()          # structured tool, builds its own SELECT

    lowered = statement.lower().lstrip("( \n\t")

    if lowered.startswith(_ORACLE_READ_PREFIXES):
        return PermissionResultAllow()
    if any(call in lowered for call in _ORACLE_READ_CALLS):
        return PermissionResultAllow()
    if _ORACLE_DICTIONARY.search(lowered) and not _WRITE_VERBS.match(lowered):
        return PermissionResultAllow()

    offending = _WRITE_VERBS.search(lowered)
    verb = offending.group(0).upper() if offending else "non-SELECT"
    return PermissionResultDeny(
        message=(
            f"Source database is read-only: {verb} rejected. "
            f"The Oracle system is live production for eleven Secretary of "
            f"State offices; the migration reads it and never writes to it. "
            f"If you need a derived value, compute it on the PostgreSQL side."
        )
    )

// hooks.ts -- @anthropic-ai/claude-agent-sdk
import { PermissionResultAllow, PermissionResultDeny } from "@anthropic-ai/claude-agent-sdk";

const READ_PREFIXES = ["select", "with"];
const READ_CALLS = ["dbms_metadata.get_ddl", "dbms_metadata.set_transform_param"];
const DICTIONARY = /\b(all|user|dba)_[a-z_]+\b/i;
const WRITE_VERBS = /\b(insert|update|delete|merge|truncate|drop|alter|create|grant|revoke|comment|flashback|lock|rename|purge|call|execute|begin|declare)\b/i;

function statementOf(input: Record<string, unknown>): string {
  for (const key of ["sql", "ddl", "statement", "query"]) {
    const value = input[key];
    if (typeof value === "string") return value;
  }
  return "";
}

export async function enforceOracleReadonly(
  toolName: string,
  toolInput: Record<string, unknown>,
) {
  if (!toolName.startsWith("mcp__oracle_src__")) return new PermissionResultAllow();

  const statement = statementOf(toolInput).trim();
  if (!statement) return new PermissionResultAllow();

  const lowered = statement.toLowerCase().replace(/^[(\s]+/, "");

  if (READ_PREFIXES.some((p) => lowered.startsWith(p))) return new PermissionResultAllow();
  if (READ_CALLS.some((c) => lowered.includes(c))) return new PermissionResultAllow();
  if (DICTIONARY.test(lowered) && !new RegExp(`^${WRITE_VERBS.source}`, "i").test(lowered)) {
    return new PermissionResultAllow();
  }

  const verb = lowered.match(WRITE_VERBS)?.[0]?.toUpperCase() ?? "non-SELECT";
  return new PermissionResultDeny({
    message:
      `Source database is read-only: ${verb} rejected. ` +
      `The Oracle system is live production for eleven Secretary of State ` +
      `offices; the migration reads it and never writes to it. If you need a ` +
      `derived value, compute it on the PostgreSQL side.`,
  });
}

Guard 3: the gate the agent cannot open

pg_cutover renames ucc_migrated to public. It is atomic, it is fast, and it is the only genuinely irreversible action in the system.

# solution/hooks.py
async def hitl_cutover_gate(tool_name: str, tool_input: dict, context):
    if tool_name != "mcp__pg_target__pg_cutover":
        return PermissionResultAllow()

    if config.CUTOVER_APPROVED:          # set ONLY by --approve-cutover on the CLI
        return PermissionResultAllow()

    summary = _read_validation_summary()  # may not exist yet; handled

    return PermissionResultDeny(
        message=(
            "CUTOVER REQUIRES HUMAN APPROVAL.\n"
            f"Current validation state: {summary}\n"
            "A person must read artifacts/validation_report.html and re-run:\n"
            "    python coordinator.py --phase cutover --approve-cutover\n"
            "Do not attempt to work around this. Report the validation state "
            "to the operator and stop."
        )
    )

Notice what this function does not do. It does not ask the model whether cutover seems reasonable. It does not check whether validation passed and approve automatically if it did. It does not offer a confidence threshold.

Self-approval is not approval. If the agent can reach a state where it proceeds, then the gate is a delay, not a control — and the whole design collapses back to "the agent decides", with extra steps. The only thing that opens this gate is a person typing a flag, which sets an environment variable this process can read and cannot write.

Animation 7: Six Checks, and the One That Cannot Be Averaged

Check 2 deserves a note, because it is the one people over-trust. A checksum fingerprint is genuinely useful for answering "did this table change since the last run?" It cannot answer "do these two databases agree", because the hash functions differ. If you find yourself comparing an ORA_HASH sum to a hashtext sum and concluding anything, stop.

The Legacy Schema

Six tables, two packages, three views, one materialized view, five sequence-and-trigger pairs. Every object is here because it contains at least one thing that does not translate cleanly.

Here is the DDL for the two most interesting tables. Read the comments — they are in the lab file too.

-- legacy-oracle/01_schema.sql
CREATE TABLE ucc_filing (
  filing_id         NUMBER(12)                         NOT NULL,
  filing_number     VARCHAR2(20 BYTE)                  NOT NULL,
  state_code        CHAR(2)                            NOT NULL,
  filing_type       VARCHAR2(12 BYTE)                  NOT NULL,
  filed_date        DATE                               NOT NULL,  -- carries TIME
  lapse_date        DATE,                                          -- carries TIME
  status            VARCHAR2(12 BYTE) DEFAULT 'ACTIVE' NOT NULL,
  collateral_desc   CLOB,
  page_count        NUMBER(4,0),
  filing_fee        NUMBER(9,2),
  created_by        VARCHAR2(30 BYTE) DEFAULT USER     NOT NULL,
  created_ts        DATE DEFAULT SYSDATE               NOT NULL,
  CONSTRAINT pk_ucc_filing PRIMARY KEY (filing_id),
  CONSTRAINT uq_ucc_filing_number UNIQUE (filing_number),
  CONSTRAINT ck_filing_type CHECK (filing_type IN
    ('UCC1','UCC3_AMD','UCC3_CONT','UCC3_TERM','UCC5'))
);

-- Function-based index. PostgreSQL supports these; the syntax differs.
CREATE INDEX ix_filing_upper_number ON ucc_filing (UPPER(filing_number));

-- legacy-oracle/01_schema.sql
-- THE PLANTED BUG LIVES HERE.
--
-- MAILING_ADDRESS_2 is populated with '' for roughly 1,400 rows. Oracle
-- stores '' as NULL. PostgreSQL stores it as a zero-length string. A
-- naive CSV round-trip turns those Oracle NULLs into PostgreSQL empty
-- strings, and every IS NULL predicate in the application silently
-- starts returning fewer rows.
CREATE TABLE ucc_debtor (
  debtor_id         NUMBER(12)                         NOT NULL,
  filing_id         NUMBER(12)                         NOT NULL,
  debtor_name       VARCHAR2(240 BYTE)                 NOT NULL,
  debtor_type       VARCHAR2(12 BYTE)                  NOT NULL,
  mailing_address_1 VARCHAR2(120 BYTE),
  mailing_address_2 VARCHAR2(120 BYTE),   -- <-- the empty-string trap
  city              VARCHAR2(60 BYTE),
  state_code        CHAR(2),
  postal_code       VARCHAR2(10 BYTE),
  CONSTRAINT pk_ucc_debtor PRIMARY KEY (debtor_id),
  CONSTRAINT fk_debtor_filing FOREIGN KEY (filing_id)
    REFERENCES ucc_filing (filing_id) ON DELETE CASCADE
);

-- legacy-oracle/02_sequences_triggers.sql
-- The pre-12c identity idiom: a sequence plus a BEFORE INSERT trigger.
-- Translation: GENERATED BY DEFAULT AS IDENTITY, plus a setval() for the
-- high-water mark after the load. The trigger itself is NOT emitted.
CREATE SEQUENCE seq_filing_id START WITH 1 INCREMENT BY 1 NOCACHE NOCYCLE;

CREATE OR REPLACE TRIGGER trg_filing_bi
BEFORE INSERT ON ucc_filing
FOR EACH ROW
WHEN (NEW.filing_id IS NULL)
BEGIN
  :NEW.filing_id := seq_filing_id.NEXTVAL;
END;
/

-- BUT: this second trigger is NOT a pure identity trigger. It also
-- normalizes the filing number and defaults a lapse date. Collapsing it
-- into an identity column would LOSE business logic. It has to survive
-- as a separate PostgreSQL trigger.
CREATE OR REPLACE TRIGGER trg_filing_normalize_bi
BEFORE INSERT ON ucc_filing
FOR EACH ROW
BEGIN
  :NEW.filing_number := UPPER(TRIM(:NEW.filing_number));
  IF :NEW.lapse_date IS NULL AND :NEW.filing_type = 'UCC1' THEN
    :NEW.lapse_date := ADD_MONTHS(:NEW.filed_date, 60);
  END IF;
END;
/

The Type Mapping, as Testable Code

A mapping table that lives only in a prompt cannot be unit-tested, cannot be diffed when someone changes it, and gives a slightly different answer on a bad day. So the mechanical part goes in type_mapping.py, where 46 assertions run in a tenth of a second, and the model spends its judgement on the cases that genuinely need it.

# solution/type_mapping.py
class Confidence(str, Enum):
    CONFIDENT = "confident"     # mechanical, no judgement needed
    CHECK_DATA = "check_data"   # the right answer depends on the values
    MANUAL = "manual"           # no good equivalent; a human decides

@dataclass(frozen=True)
class Mapping:
    oracle_type: str
    postgres_type: str
    reason: str
    confidence: Confidence = Confidence.CONFIDENT

    @property
    def needs_review(self) -> bool:
        return self.confidence is not Confidence.CONFIDENT


def map_type(oracle_type: str, *, sample_values: list | None = None) -> Mapping:
    upper = (oracle_type or "").strip().upper()

    if upper == "DATE":
        return Mapping(
            oracle_type, "timestamp(0)",
            "Oracle DATE carries a TIME component. Mapping it to `date` "
            "silently truncates 14:32:07 to midnight, which changes which "
            "rows appear to have lapsed.",
        )

    raw = _RAW.match(upper)
    if raw and int(raw.group(1)) == 16:
        looks_like_uuid = bool(sample_values) and all(
            isinstance(v, str) and len(v.replace("-", "")) == 32
            for v in sample_values if v is not None
        )
        if looks_like_uuid:
            return Mapping(oracle_type, "uuid",
                           "RAW(16) holding SYS_GUID values is a uuid.")
        return Mapping(
            oracle_type, "uuid",
            "RAW(16) is usually SYS_GUID -- but confirm against real rows. "
            "If the bytes are a hash rather than a GUID, use bytea.",
            Confidence.CHECK_DATA,
        )

    # ... NUMBER precision rules, VARCHAR2 BYTE/CHAR, TIMESTAMP variants ...

    return Mapping(oracle_type, "text",
                   f"No mapping rule for {oracle_type!r}. Defaulting to text "
                   f"is a placeholder, not an answer -- a human must decide.",
                   Confidence.MANUAL)

# tests/test_type_mapping.py
def test_oracle_date_becomes_timestamp_not_date():
    """The single most consequential row in the mapping table.

    Oracle DATE carries a time component. Mapping it to `date` compiles,
    loads, and silently truncates every filed_date to midnight -- which
    changes which filings look lapsed. Nothing errors.
    """
    mapping = map_type("DATE")
    assert mapping.postgres_type == "timestamp(0)"
    assert mapping.postgres_type != "date"
    assert "time component" in mapping.reason.lower()


def test_raw16_without_samples_still_says_uuid_but_asks_to_check():
    """The DDL alone cannot distinguish a GUID from a 16-byte hash. The
    mapping guesses the common case and says it is guessing."""
    mapping = map_type("RAW(16)")
    assert mapping.postgres_type == "uuid"
    assert mapping.needs_review


def test_identifiers_are_lowercased_not_quoted():
    """Oracle folds unquoted names to upper, PostgreSQL to lower. Quoting
    to preserve the uppercase name would force every query afterwards to
    quote it too."""
    assert quote_policy("UCC_FILING") == "ucc_filing"
    assert '"' not in quote_policy("UCC_FILING")


@pytest.mark.parametrize("column,oracle,expected", [
    ("UCC_FILING.FILING_ID",            "NUMBER(12)",         "bigint"),
    ("UCC_FILING.FILED_DATE",           "DATE",               "timestamp(0)"),
    ("UCC_FILING.FILING_FEE",           "NUMBER(9,2)",        "numeric(9,2)"),
    ("UCC_SECURED_PARTY.TAX_ID",        "RAW(16)",            "uuid"),
    ("STATE_SOS_SOURCE.RECORDS_EXPECTED", "NUMBER",           "numeric"),
    ("FILING_AUDIT.DOC_IMAGE",          "BLOB",               "bytea"),
])
def test_every_column_in_the_legacy_schema(column, oracle, expected):
    assert map_type(oracle).postgres_type == expected, column

File Structure

Everything that is not agent logic ships complete. You build the guardrails, the mapping, the reconciliation, the orchestration, and the five subagent definitions.

Spec-Driven: Build It Twice

This is a Tier 3 capstone, which means spec/agent-spec.md is not documentation — it is the source. Once your hand-built version works, you build the whole thing again from the spec in one command, and then diff.

# After your hand-built version passes the tests:
claude "Read spec/agent-spec.md and build the entire project into generated/"

# Then compare
diff -ru solution/ generated/

The spec has twelve sections — business context, agent configuration, tools, subagents, hooks, guardrails, sessions, deployment, observability, tests, evaluation dataset, file structure — and it ends with acceptance criteria that are checkable rather than aspirational:

## Acceptance Criteria

- All Python files import from `claude_agent_sdk` only -- no
  `client.messages.create()` anywhere outside `appendix/manual-loop.py`
- `pytest tests/ -v` passes 100%
- `python evaluation/test_suite.py` scores >= 18/20
- `docker compose up` brings up both databases and the agent completes
  phases 1-5 unattended, then **stops** at the cutover gate
- Every row of `UCC_DEBTOR` that was NULL in Oracle is NULL -- not empty
  string -- in PostgreSQL
- `migration_audit.jsonl` has one entry per tool call, with no credentials
- Attempting any write against Oracle is denied and logged

Step-by-Step Build Guide

Environment setup (one block, copy-paste it all)

cd labs/capstone-8-oracle-to-postgres

# Credentials
cp starter/.env.example starter/.env
# Edit starter/.env and add your ANTHROPIC_API_KEY

# Dependencies, for running tests on the host
pip install -r requirements.txt

# Bring up both databases
cd starter
docker compose up -d oracle postgres

# WAIT. This is the step people skip.
docker compose ps
# oracle must read "healthy", not just "running"
docker compose logs -f oracle
# You are waiting for: DATABASE IS READY TO USE!

# Confirm the legacy schema seeded
docker compose exec oracle sqlplus -S migration_reader/'ReadOnly#2026'@FREEPDB1 \
  <<< "select count(*) from meridian.ucc_filing;"

# starter/hooks.py
_ORACLE_READ_PREFIXES = ("select", "with")
_ORACLE_READ_CALLS = ("dbms_metadata.get_ddl", "dbms_metadata.set_transform_param")
_ORACLE_DICTIONARY = re.compile(r"\b(all|user|dba)_[a-z_]+\b", re.I)

# Used ONLY to name the offending verb in the denial message.
# It is not the security boundary -- the allow-list above is.
_WRITE_VERBS = re.compile(
    r"\b(insert|update|delete|merge|truncate|drop|alter|create|grant|revoke|"
    r"comment|flashback|lock|rename|purge|call|execute|begin|declare)\b",
    re.I,
)


async def enforce_oracle_readonly(tool_name: str, tool_input: dict, context: Any):
    # (1b) Not an oracle_src tool? Not our business.
    #      Omit this and the guard blocks every PostgreSQL write too, and
    #      the migration does nothing while looking well protected.
    if not tool_name.startswith("mcp__oracle_src__"):
        return PermissionResultAllow()

    # (1c) Structured tools carry no free-form SQL -- they build their own
    #      SELECT from a validated identifier. Nothing to inspect.
    statement = _statement_of(tool_input).strip()
    if not statement:
        return PermissionResultAllow()

    lowered = statement.lower().lstrip("( \n\t")

    # (1d) The three permitted shapes.
    if lowered.startswith(_ORACLE_READ_PREFIXES):
        return PermissionResultAllow()
    if any(call in lowered for call in _ORACLE_READ_CALLS):
        return PermissionResultAllow()
    if _ORACLE_DICTIONARY.search(lowered) and not _WRITE_VERBS.match(lowered):
        return PermissionResultAllow()

    # (1e) Everything else. Name the verb -- a bare "denied" makes the
    #      model rephrase and retry, and you pay for the loop.
    offending = _WRITE_VERBS.search(lowered)
    verb = offending.group(0).upper() if offending else "non-SELECT"
    return PermissionResultDeny(
        message=(
            f"Source database is read-only: {verb} rejected. "
            f"The Oracle system is live production for eleven Secretary of "
            f"State offices; the migration reads it and never writes to it. "
            f"If you need a derived value, compute it on the PostgreSQL side."
        )
    )

cd labs/capstone-8-oracle-to-postgres
TEST_TARGET=starter pytest tests/test_hooks_readonly.py -v

# starter/hooks.py
_PG_FORBIDDEN = re.compile(
    r"\b(drop\s+database|drop\s+schema|drop\s+owned|drop\s+role|"
    r"drop\s+tablespace)\b",
    re.I,
)
_PG_CREATES = re.compile(
    r"\b(create|alter)\s+(table|view|materialized\s+view|index|sequence|"
    r"function|procedure|type|trigger)\s+(if\s+not\s+exists\s+)?([a-z0-9_\".]+)",
    re.I,
)


async def protect_pg_target(tool_name: str, tool_input: dict, context: Any):
    if tool_name != "mcp__pg_target__pg_apply_ddl":
        return PermissionResultAllow()

    ddl = (tool_input.get("ddl") or "").strip()
    if not ddl:
        return PermissionResultAllow()

    forbidden = _PG_FORBIDDEN.search(ddl)
    if forbidden:
        return PermissionResultDeny(
            message=(
                f"Refused: '{forbidden.group(0)}' is never part of a migration. "
                f"Objects are created inside {config.POSTGRES.target_schema} and "
                f"the schema is promoted at cutover; nothing is ever dropped."
            )
        )

    # Unqualified names are fine -- search_path puts them in the target.
    # A qualified name pointing anywhere else is not.
    for match in _PG_CREATES.finditer(ddl):
        name = match.group(4).strip('"')
        if "." in name:
            schema = name.split(".", 1)[0].strip('"').lower()
            if schema != config.POSTGRES.target_schema:
                return PermissionResultDeny(
                    message=(
                        f"Refused: '{name}' targets schema '{schema}'. Every "
                        f"migrated object must be created in "
                        f"'{config.POSTGRES.target_schema}' so the cutover is a "
                        f"single atomic schema rename."
                    )
                )

    return PermissionResultAllow()

TEST_TARGET=starter pytest tests/test_hooks_pg_guard.py -v

# starter/hooks.py
async def hitl_cutover_gate(tool_name: str, tool_input: dict, context: Any):
    if tool_name != "mcp__pg_target__pg_cutover":
        return PermissionResultAllow()

    if config.CUTOVER_APPROVED:
        return PermissionResultAllow()

    # Carry the validation state into the denial. An operator who has to go
    # find the numbers themselves will eventually stop reading them.
    summary_path = os.path.join(config.ARTIFACT_DIR, "validation_summary.json")
    summary = "no validation report found -- run phase 5 first"
    if os.path.exists(summary_path):
        try:
            with open(summary_path, encoding="utf-8") as fh:
                data = json.load(fh)
            summary = (
                f"{data.get('tables_validated', '?')} tables validated, "
                f"{data.get('checks_passed', '?')} checks passed, "
                f"{data.get('checks_failed', '?')} failed"
            )
        except (OSError, json.JSONDecodeError):
            summary = "validation report present but unreadable"

    return PermissionResultDeny(
        message=(
            "CUTOVER REQUIRES HUMAN APPROVAL.\n"
            f"Current validation state: {summary}\n"
            "A person must read artifacts/validation_report.html and re-run:\n"
            "    python coordinator.py --phase cutover --approve-cutover\n"
            "Do not attempt to work around this. Report the validation state "
            "to the operator and stop."
        )
    )


# --- the audit log, TODO(4a)-(4e) -------------------------------------
_SECRET_PATTERNS = [
    re.compile(r"(password\s*=\s*)(\S+)", re.I),
    re.compile(r"(ORACLE_PWD\s*=\s*)(\S+)", re.I),
    re.compile(r"(://[^:/\s]+:)([^@/\s]+)(@)"),        # user:pass@host
    re.compile(r"(\buser\s*=\s*\S+\s+password\s*=\s*)(\S+)", re.I),
    re.compile(r"(sk-ant-[A-Za-z0-9_-]{6})([A-Za-z0-9_-]+)"),
]


def redact(text: str) -> str:
    """Deliberately aggressive. An audit log that over-redacts is annoying;
    one that leaks a production DSN into a committed file is an incident."""
    for pattern in _SECRET_PATTERNS:
        if pattern.groups >= 3:
            text = pattern.sub(r"\1***\3", text)
        else:
            text = pattern.sub(r"\1***", text)
    return text

TEST_TARGET=starter pytest tests/test_cutover_hitl.py -v

TEST_TARGET=starter pytest tests/test_type_mapping.py -v --tb=short

---
name: plsql-converter
description: Converts one PL/SQL package, procedure, function, or trigger to
  PL/pgSQL, or refuses and explains why. Use for phase 4 of the migration.
tools: mcp__oracle_src__oracle_get_plsql_source, mcp__pg_target__pg_apply_ddl,
  mcp__migration_local__write_artifact
model: claude-sonnet-4-6
---

You convert one PL/SQL object to PL/pgSQL.

## Packages

PostgreSQL has no packages. Convert a package to a **schema** of the same
name containing one function per public routine. This matters downstream:
application code calling `pkg_risk_calc.score_debtor(x)` keeps working
unchanged, but only because you created a schema with that exact name.
Record that in the decision log, so the next person understands why a
schema is named after a package.

## When to refuse

**`PRAGMA AUTONOMOUS_TRANSACTION` has no safe PostgreSQL equivalent.**

The whole point of the pragma is that the row commits even when the
calling transaction rolls back. That is exactly what you want from an
audit log. PostgreSQL cannot do it in-process: the options are `dblink`,
a background worker, or moving the write out of the transaction entirely
at the application layer. All three are design decisions, not
translations.

So do not translate it. And in particular, do not drop the pragma and
emit the rest -- that compiles, it runs, and it silently inverts the
semantics, so audit rows start vanishing on exactly the rollbacks you
most wanted them for.

Refuse. Write the analysis to `plsql/<object>.MANUAL_REVIEW.md` explaining
what the pragma does, why it cannot be translated, and what each of the
three redesign options costs. Then report it as queued for manual review.

The same applies to anything else you are not confident in. A refusal with
a reason is a useful output. A confident wrong translation is not.

TEST_TARGET=starter pytest tests/test_plsql_conversion.py -v

# starter/coordinator.py
from claude_agent_sdk import (
    AssistantMessage, ClaudeAgentOptions, HookMatcher, query,
)

def _options(system_prompt: str, model: str, allowed: list[str] | None = None):
    """One place where every SDK knob is set.

    Centralised deliberately: a second construction site is a second
    chance to forget can_use_tool, and a migration with no guardrails
    looks exactly like a migration with guardrails right up until it
    does not.
    """
    return ClaudeAgentOptions(
        model=model,
        system_prompt=system_prompt,
        max_turns=config.MAX_TURNS,
        mcp_servers={
            "oracle_src": oracle_server,
            "pg_target": pg_server,
            "migration_local": local_server,
        },
        allowed_tools=allowed,
        can_use_tool=hooks.can_use_tool,                       # the 3 guards
        hooks=[HookMatcher(matcher="*", hooks=[hooks.audit_log])],
    )


async def _run(prompt, *, system_prompt, model, tracer, label, budget) -> str:
    """Run one agent turn, trace it, and charge the budget."""
    chunks: list[str] = []
    with tracer.span(label) as span:
        async for message in query(prompt=prompt,
                                   options=_options(system_prompt, model)):
            if isinstance(message, AssistantMessage):
                for block in message.content:
                    text = getattr(block, "text", None)
                    if text:
                        chunks.append(text)
        usage = getattr(message, "usage", None)
        output_tokens = getattr(usage, "output_tokens", 0) if usage else 0
        budget.add(output_tokens)
        span.tokens = output_tokens
    return "\n".join(chunks)

// coordinator.ts -- @anthropic-ai/claude-agent-sdk
import { query, type Options } from "@anthropic-ai/claude-agent-sdk";
import { canUseTool, auditLog, type TokenBudget } from "./hooks.js";
import { oracleServer } from "./toolsOracle.js";
import { pgServer } from "./toolsPostgres.js";
import { localServer } from "./toolsLocal.js";
import * as config from "./config.js";

function options(systemPrompt: string, model: string): Options {
  return {
    model,
    systemPrompt,
    maxTurns: config.MAX_TURNS,
    mcpServers: {
      oracle_src: oracleServer,
      pg_target: pgServer,
      migration_local: localServer,
    },
    canUseTool,                                   // the 3 guards
    hooks: [{ matcher: "*", hooks: [auditLog] }],
  };
}

export async function run(
  prompt: string,
  opts: { systemPrompt: string; model: string; tracer: Tracer; label: string; budget: TokenBudget },
): Promise<string> {
  const chunks: string[] = [];
  const span = opts.tracer.start(opts.label);
  try {
    for await (const message of query({
      prompt,
      options: options(opts.systemPrompt, opts.model),
    })) {
      if (message.type === "assistant") {
        for (const block of message.content) {
          if (block.type === "text") chunks.push(block.text);
        }
      }
      if (message.usage?.output_tokens) {
        opts.budget.add(message.usage.output_tokens);
        span.tokens += message.usage.output_tokens;
      }
    }
  } finally {
    span.end();
  }
  return chunks.join("\n");
}

cd starter
docker compose run --rm agent python coordinator.py --phase discover

# starter/validation.py
def detect_empty_string_divergence(
    table: str, columns: list[str], oracle: dict, postgres: dict
) -> list[Defect]:
    """Oracle stores '' as NULL. PostgreSQL stores it as a zero-length
    string. If a column had NULLs in Oracle and now has empty strings in
    PostgreSQL, those NULLs were converted on the way across."""
    defects: list[Defect] = []
    for column in columns:
        empties = int(postgres.get(f"empty_{column.lower()}", 0) or 0)
        if empties == 0:
            continue

        oracle_nulls = int(oracle.get(f"null_{column.lower()}", 0) or 0)
        pg_nulls = int(postgres.get(f"null_{column.lower()}", 0) or 0)

        if oracle_nulls > 0 and pg_nulls < oracle_nulls:
            # The bug. Oracle had NULLs; PostgreSQL has fewer, and has
            # empty strings instead. Those are the same rows.
            defects.append(Defect(
                check="empty_string_divergence",
                object=f"{table}.{column}",
                detail=(
                    f"{empties:,} empty strings in PostgreSQL. Oracle "
                    f"reported {oracle_nulls:,} NULLs here; PostgreSQL "
                    f"reports only {pg_nulls:,}. "
                    f"{oracle_nulls - pg_nulls:,} Oracle NULLs were "
                    f"converted to empty strings by the load. Every "
                    f"IS NULL query against this column now returns "
                    f"fewer rows than it did. Re-load with null_as set."
                ),
                severity=Severity.BLOCKER,
            ))
        else:
            # Empty strings, but the NULL counts reconcile. Probably
            # genuine. Flag it; do not block on it.
            defects.append(Defect(
                check="empty_string_present",
                object=f"{table}.{column}",
                detail=(
                    f"{empties:,} empty strings in PostgreSQL, but the "
                    f"Oracle NULL count ({oracle_nulls:,}) is consistent "
                    f"with the PostgreSQL NULL count ({pg_nulls:,}). "
                    f"Probably genuine empty strings rather than "
                    f"converted NULLs -- confirm before dismissing."
                ),
                severity=Severity.WARNING,
            ))
    return defects

TEST_TARGET=starter pytest tests/test_validator_catches_empty_string.py -v

cd starter
docker compose run --rm agent python coordinator.py --migrate-all

docker compose run --rm agent python coordinator.py --phase data
docker compose run --rm agent python coordinator.py --phase validate

docker compose run --rm agent python coordinator.py --phase cutover

# 1. The full unit suite -- no database, no API key
TEST_TARGET=starter pytest tests/ -v

# 2. Guardrails actually block, not just in tests
docker compose run --rm agent python -c "
import asyncio, hooks
print(asyncio.run(hooks.can_use_tool(
    'mcp__oracle_src__oracle_sample_rows',
    {'sql': 'DROP TABLE meridian.ucc_filing'}, None)).message)"

# 3. Nothing leaked into the audit log
grep -ci password migration_audit.jsonl

# 4. The evaluation harness
docker compose run --rm agent python evaluation/test_suite.py

Test Scenarios

Twenty scenarios in evaluation/test_cases.json, across five categories. Here are the ten that carry the most information about whether your agent works.

Tier 1: Local Production Deployment

No cloud account. Two databases and the agent, in Docker Compose.

services:
  oracle:
    image: gvenzl/oracle-free:23-slim
    platform: linux/amd64          # no ARM build; emulated on Apple Silicon
    environment:
      ORACLE_PASSWORD: ${ORACLE_SYS_PASSWORD:-MeridianSys#2003}
      APP_USER: meridian
      APP_USER_PASSWORD: ${ORACLE_APP_PASSWORD:-MeridianLegacy#2003}
    volumes:
      # Scripts run once, in lexical order, on FIRST BOOT ONLY.
      # Re-seed with: docker compose down -v
      - ../legacy-oracle:/container-entrypoint-initdb.d:ro
      - oracle-data:/opt/oracle/oradata
    healthcheck:
      test: ["CMD", "healthcheck.sh"]
      interval: 15s
      timeout: 10s
      retries: 40                  # 40 x 15s = 10 min ceiling for a cold start
      start_period: 60s

  postgres:
    image: postgres:16-alpine
    environment:
      POSTGRES_DB: meridian
      POSTGRES_USER: migration
      POSTGRES_PASSWORD: ${PG_PASSWORD:-migration}
      POSTGRES_INITDB_ARGS: "--encoding=UTF8 --locale=C"
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U migration -d meridian"]
      interval: 5s
      retries: 20

  agent:
    build: .
    depends_on:
      oracle:   { condition: service_healthy }   # gate on health, not a timer
      postgres: { condition: service_healthy }
    environment:
      ANTHROPIC_API_KEY: ${ANTHROPIC_API_KEY:?ANTHROPIC_API_KEY is required}
      ORACLE_USER: migration_reader              # SELECT and nothing else
      ORACLE_DSN: oracle:1521/FREEPDB1
      PG_HOST: postgres
      PG_TARGET_SCHEMA: ucc_migrated
      # Deliberately NOT set. Cutover requires a human to pass
      # --approve-cutover, which the agent cannot do for itself.
      # CUTOVER_APPROVED:
    command: ["python", "coordinator.py", "--migrate-all"]

Three details in that file are load-bearing.

condition: service_healthy, not a sleep. Oracle takes 1–3 minutes on a cold boot, and longer under emulation. sleep 90 works on your laptop and fails in CI, which is the worst place to discover it.

The init scripts run on first boot only. Edit legacy-oracle/*.sql after the volume exists and nothing re-runs — you will get a confusing ORA-01017 and assume the credentials are wrong. docker compose down -v is the fix.

CUTOVER_APPROVED is commented out, with a comment saying why. A future maintainer looking for it will find the explanation instead of the variable, which is the entire purpose of writing it down.

Cloud Deployment (Optional)

Both cloud tiers need an account. Skip them if you do not have one — the lab is complete without them.

Why a job, not a service

Phases 1–5 run for minutes to hours and then finish. That is a job, not a request handler. Deploying it behind an HTTP endpoint means fighting a request timeout for no benefit — Cloud Run services cap at 60 minutes, Lambda at 15, and a real migration of a real database exceeds both. Cloud Run jobs and ECS tasks have no such ceiling.

The cutover gate makes this matter more than it usually would. A job that dies at minute 59 leaves you resuming from session_state.json, which is fine. A job that dies during an ALTER SCHEMA RENAME does not — which is one more reason cutover is a separate, human-triggered invocation rather than the last step of a long run.

# deploy/gcp/deploy.sh -- abridged

# The migration job. Note what is NOT here: CUTOVER_APPROVED.
gcloud run jobs deploy ucc-migration-agent \
  --image "${IMAGE}" \
  --task-timeout 3600s \
  --max-retries 0 \
  --set-cloudsql-instances "${PROJECT}:${REGION}:${PG_INSTANCE}" \
  --set-secrets "ANTHROPIC_API_KEY=anthropic-api-key:latest,\
ORACLE_PASSWORD=oracle-reader-password:latest,\
PG_PASSWORD=pg-migration-password:latest" \
  --command python \
  --args coordinator.py,--migrate-all

# The CUTOVER job. Separate deployment, separate invocation, separate
# audit trail. The point is not technical -- it is that "run the
# migration" and "promote it to production" become two things a person
# decides to do, minutes or days apart, with the report in between.
gcloud run jobs deploy ucc-migration-agent-cutover \
  --image "${IMAGE}" \
  --task-timeout 600s \
  --set-secrets "CUTOVER_TOKEN=cutover-token:latest,..." \
  --command python \
  --args coordinator.py,--phase,cutover,--approve-cutover

What changes between tiers is almost nothing in the agent — config.py reads everything from the environment. The real differences are credentials (.env vs Secret Manager vs Secrets Manager), the network path to a production Oracle that is usually not reachable from a managed runtime, artifact storage (a bind mount locally, GCS or S3 in the cloud, or the generated DDL dies with the container), and how the approval gate is expressed.

Test Cases

Roughly 150 assertions across six files. None of them need a database or an API key, which is the point — a guardrail you can only exercise against a live production system is a guardrail nobody exercises.

# Against the reference implementation
pytest tests/ -v

# Against your own work -- same suite, same meaning
TEST_TARGET=starter pytest tests/ -v

# One area while you are building it
TEST_TARGET=starter pytest tests/test_hooks_readonly.py -v

📂 Get the files: labs/capstone-8-oracle-to-postgres on GitHub — or clone the course once: git clone https://github.com/varasrinivas/agenticai-course.git

Troubleshooting

Going Further [ALL OPTIONAL]

You have finished the capstone. Everything below is extra credit — skipping all of it costs you nothing.

1. A performance-advisor subagent. Reads Oracle execution plans and index definitions, proposes PostgreSQL indexes. Do this one through the spec: add a paragraph to agent-spec.md, regenerate, and compare against what adding it by hand would have cost you in six files.

2. Emit a Flyway or Liquibase changeset instead of raw DDL, so the migration is replayable and versioned. Interesting mostly for what it forces you to decide: what is a migration step when the generator is a model, and what happens when a regeneration produces a different but equivalent step?

3. Change-data-capture for a zero-downtime cutover. Logical replication keeping PostgreSQL current while Oracle still serves traffic, so the cutover window is seconds instead of hours. This is where real migrations spend their engineering budget, and it makes the HITL gate more interesting rather than less — you now have a live tail to reason about.

4. Point the same spec at MySQL. Change the target in agent-spec.md, regenerate, and read what the agent produces. The empty-string trap disappears; a raft of new ones (implicit type coercion, ONLY_FULL_GROUP_BY, zero dates) appear. Good for calibrating how much of what you learned was about Oracle and how much was about migration.

5. Wire the migration report into Grafana alongside the M19/M20 dashboards, so a long-running migration is observable while it runs rather than only afterwards.

References & Resources

This capstone

• labs/capstone-8-oracle-to-postgres/README.md — setup, build order, verification
• labs/capstone-8-oracle-to-postgres/spec/agent-spec.md — the twelve-section spec that regenerates the whole project
• labs/capstone-8-oracle-to-postgres/expected_output/ — what a correct run looks like, including the first-run failure
• prompts/19-sdk-tier-policy.md — why this is Tier 3 and what that requires
• prompts/17-spec-driven-development.md — the spec-driven pattern

Course modules

• M07 — Model Context Protocol
• M14 — Multi-Agent Systems
• M15B — Build an Agent + Subagent System
• M16 — Input Guardrails
• M17 — Output Guardrails & Human-in-the-Loop
• M18 — Evaluation & Testing
• M22B — Deploy Your Agent
• Capstone 6 — Parallel State Testing Agent
• Capstone 7-C — Agent Evolution

External

• Claude Agent SDK — overview
• Subagents in .claude/agents/
• Hooks and .claude/settings.json
• PostgreSQL 16 — data types
• PostgreSQL 16 — PL/pgSQL
• PostgreSQL 16 — WITH RECURSIVE
• gvenzl/oracle-free container image
• python-oracledb — thin mode, no Instant Client