diff --git a/hypaware-core/plugins-workspace/ai-gateway-graph/src/graph_contract.js b/hypaware-core/plugins-workspace/ai-gateway-graph/src/graph_contract.js
index e60af86..b4caa45 100644
--- a/hypaware-core/plugins-workspace/ai-gateway-graph/src/graph_contract.js
+++ b/hypaware-core/plugins-workspace/ai-gateway-graph/src/graph_contract.js
@@ -33,11 +33,13 @@ const FILE_TOOLS = new Set(['Read', 'Edit', 'Write', 'MultiEdit', 'NotebookEdit'
  * hand-authored node/edge mappings that used to live in `@hypaware/context-graph`;
  * they now live here, beside the source they read. Rows are built with the
  * graph plugin's `kit` so the id recipe and provenance columns stay owned by
- * the graph plugin. This connector owns the SQL + `toRow` semantics and the
- * bridge-key recipe (`keys`, imported from `./graph-keys.js`).
+ * the graph plugin. This connector owns the read declarations (declarative
+ * columns/where, or raw SQL for the pushed-down prefix guards) + `toRow`
+ * semantics and the bridge-key recipe (`keys`, imported from
+ * `./graph-keys.js`).
  *
  * @param {GraphKit} kit
- * @returns {{ name: string, plugin: string, sourceDataset: string, projector: string, projectorVersion: number, rules: ContractRule[] }}
+ * @returns {{ name: string, plugin: string, sourceDataset: string, projector: string, projectorVersion: number, rules: ContractRule[], rowFilter: { columns: string[], keep(row: Record<string, unknown>): boolean } }}
  * @ref LLP 0023#contract-contribution [implements]: a source's contract, contributed via the capability; engine + kit stay central
  */
 export function createAiGatewayGraphContract(kit) {
@@ -58,7 +60,7 @@ export function createAiGatewayGraphContract(kit) {
     {
       kind: 'node',
       type: 'Session',
-      sql: `SELECT session_id, cwd, git_branch, client_name, user_id, message_created_at FROM ${SOURCE_DATASET}`,
+      columns: ['session_id', 'cwd', 'git_branch', 'client_name', 'user_id', 'message_created_at'],
       toRow(r) {
         const key = str(r.session_id)
         if (!key) return null
@@ -81,7 +83,7 @@ export function createAiGatewayGraphContract(kit) {
     {
       kind: 'node',
       type: 'App',
-      sql: `SELECT client_name, message_created_at FROM ${SOURCE_DATASET}`,
+      columns: ['client_name', 'message_created_at'],
       toRow(r) {
         const key = str(r.client_name)
         if (!key) return null
@@ -93,7 +95,7 @@ export function createAiGatewayGraphContract(kit) {
     {
       kind: 'node',
       type: 'Model',
-      sql: `SELECT model, message_created_at FROM ${SOURCE_DATASET}`,
+      columns: ['model', 'message_created_at'],
       toRow(r) {
         const key = str(r.model)
         if (!key) return null
@@ -105,7 +107,8 @@ export function createAiGatewayGraphContract(kit) {
     {
       kind: 'node',
       type: 'Tool',
-      sql: `SELECT tool_name, message_created_at FROM ${SOURCE_DATASET} WHERE part_type = 'tool_call'`,
+      columns: ['tool_name', 'message_created_at'],
+      where: { eq: { part_type: 'tool_call' } },
       toRow(r) {
         const key = str(r.tool_name)
         if (!key) return null
@@ -122,7 +125,8 @@ export function createAiGatewayGraphContract(kit) {
     {
       kind: 'node',
       type: 'File',
-      sql: `SELECT tool_name, tool_args, git_remote, repo_root, message_created_at FROM ${SOURCE_DATASET} WHERE part_type = 'tool_call'`,
+      columns: ['tool_name', 'tool_args', 'git_remote', 'repo_root', 'message_created_at'],
+      where: { eq: { part_type: 'tool_call' } },
       toRow(r) {
         const target = fileTargetFrom(keys, r.tool_name, r.tool_args, r.git_remote, r.repo_root)
         if (!target) return null
@@ -135,7 +139,7 @@ export function createAiGatewayGraphContract(kit) {
     {
       kind: 'node',
       type: 'Repo',
-      sql: `SELECT git_remote, message_created_at FROM ${SOURCE_DATASET}`,
+      columns: ['git_remote', 'message_created_at'],
       toRow(r) {
         const key = keys.repoKeyFromRemote(r.git_remote)
         if (!key) return null
@@ -150,7 +154,7 @@ export function createAiGatewayGraphContract(kit) {
     {
       kind: 'node',
       type: 'Commit',
-      sql: `SELECT head_sha, message_created_at FROM ${SOURCE_DATASET}`,
+      columns: ['head_sha', 'message_created_at'],
       toRow(r) {
         const key = keys.commitKey(r.head_sha)
         if (!key) return null
@@ -168,7 +172,8 @@ export function createAiGatewayGraphContract(kit) {
     {
       kind: 'node',
       type: 'Program',
-      sql: `SELECT tool_name, tool_args, message_created_at FROM ${SOURCE_DATASET} WHERE part_type = 'tool_call' AND tool_name IN ('Bash', 'exec_command')`,
+      columns: ['tool_name', 'tool_args', 'message_created_at'],
+      where: { eq: { part_type: 'tool_call' }, in: { tool_name: ['Bash', 'exec_command'] } },
       toRow(r) {
         const key = programFrom(commandStringFrom(r.tool_name, r.tool_args))
         if (!key) return null
@@ -193,7 +198,8 @@ export function createAiGatewayGraphContract(kit) {
     {
       kind: 'node',
       type: 'Skill',
-      sql: `SELECT session_id, tool_args, message_created_at FROM ${SOURCE_DATASET} WHERE part_type = 'tool_call' AND tool_name = 'Skill'`,
+      columns: ['session_id', 'tool_args', 'message_created_at'],
+      where: { eq: { part_type: 'tool_call', tool_name: 'Skill' } },
       toRow(r) {
         const key = skillFromToolArgs(r.tool_args)
         if (!key) return null
@@ -205,7 +211,12 @@ export function createAiGatewayGraphContract(kit) {
     {
       kind: 'node',
       type: 'Skill',
-      sql: `SELECT session_id, content_text, message_created_at FROM ${SOURCE_DATASET} WHERE role = 'user' AND part_type = 'text' AND content_text LIKE 'Base directory for this skill: %'`,
+      // Raw SQL on purpose: the prefix filter prunes `content_text` (the
+      // table's largest column) server-side, keeping it out of the shared
+      // scan's materialization. `attributes` rides for the contract
+      // rowFilter (registry-enforced).
+      // @ref LLP 0096#decision [constrained-by]: heavy-column prefix guards stay pushed down
+      sql: `SELECT attributes, session_id, content_text, message_created_at FROM ${SOURCE_DATASET} WHERE role = 'user' AND part_type = 'text' AND content_text LIKE 'Base directory for this skill: %'`,
       toRow(r) {
         const key = skillFromMarker(r.content_text)
         if (!key) return null
@@ -217,7 +228,8 @@ export function createAiGatewayGraphContract(kit) {
     {
       kind: 'node',
       type: 'Skill',
-      sql: `SELECT session_id, content_text, message_created_at FROM ${SOURCE_DATASET} WHERE role = 'user' AND part_type = 'text' AND content_text LIKE '<command-name>%'`,
+      // Raw SQL on purpose, same pushdown rationale as the marker surface.
+      sql: `SELECT attributes, session_id, content_text, message_created_at FROM ${SOURCE_DATASET} WHERE role = 'user' AND part_type = 'text' AND content_text LIKE '<command-name>%'`,
       toRow(r) {
         const key = skillFromSlash(r.content_text)
         if (!key) return null
@@ -234,7 +246,8 @@ export function createAiGatewayGraphContract(kit) {
     {
       kind: 'node',
       type: 'Skill',
-      sql: `SELECT session_id, tool_args, message_created_at FROM ${SOURCE_DATASET} WHERE part_type = 'tool_call' AND tool_name = 'exec_command'`,
+      columns: ['session_id', 'tool_args', 'message_created_at'],
+      where: { eq: { part_type: 'tool_call', tool_name: 'exec_command' } },
       toRow(r) {
         const key = skillFromCodexRead(commandStringFrom('exec_command', r.tool_args))
         if (!key) return null
@@ -249,7 +262,7 @@ export function createAiGatewayGraphContract(kit) {
     {
       kind: 'edge',
       type: 'via',
-      sql: `SELECT session_id, client_name, message_created_at FROM ${SOURCE_DATASET}`,
+      columns: ['session_id', 'client_name', 'message_created_at'],
       toRow(r) {
         const session = str(r.session_id)
         const app = str(r.client_name)
@@ -262,7 +275,7 @@ export function createAiGatewayGraphContract(kit) {
     {
       kind: 'edge',
       type: 'used_model',
-      sql: `SELECT session_id, model, message_created_at FROM ${SOURCE_DATASET}`,
+      columns: ['session_id', 'model', 'message_created_at'],
       toRow(r) {
         const session = str(r.session_id)
         const model = str(r.model)
@@ -275,7 +288,8 @@ export function createAiGatewayGraphContract(kit) {
     {
       kind: 'edge',
       type: 'used',
-      sql: `SELECT session_id, tool_name, message_created_at FROM ${SOURCE_DATASET} WHERE part_type = 'tool_call'`,
+      columns: ['session_id', 'tool_name', 'message_created_at'],
+      where: { eq: { part_type: 'tool_call' } },
       toRow(r) {
         const session = str(r.session_id)
         const tool = str(r.tool_name)
@@ -290,7 +304,8 @@ export function createAiGatewayGraphContract(kit) {
     {
       kind: 'edge',
       type: 'touched',
-      sql: `SELECT session_id, tool_name, tool_args, git_remote, repo_root, message_created_at FROM ${SOURCE_DATASET} WHERE part_type = 'tool_call'`,
+      columns: ['session_id', 'tool_name', 'tool_args', 'git_remote', 'repo_root', 'message_created_at'],
+      where: { eq: { part_type: 'tool_call' } },
       toRow(r) {
         const session = str(r.session_id)
         const target = fileTargetFrom(keys, r.tool_name, r.tool_args, r.git_remote, r.repo_root)
@@ -303,7 +318,7 @@ export function createAiGatewayGraphContract(kit) {
     {
       kind: 'edge',
       type: 'in',
-      sql: `SELECT session_id, git_remote, message_created_at FROM ${SOURCE_DATASET}`,
+      columns: ['session_id', 'git_remote', 'message_created_at'],
       toRow(r) {
         const session = str(r.session_id)
         const repo = keys.repoKeyFromRemote(r.git_remote)
@@ -316,7 +331,7 @@ export function createAiGatewayGraphContract(kit) {
     {
       kind: 'edge',
       type: 'at',
-      sql: `SELECT session_id, head_sha, message_created_at FROM ${SOURCE_DATASET}`,
+      columns: ['session_id', 'head_sha', 'message_created_at'],
       toRow(r) {
         const session = str(r.session_id)
         const commit = keys.commitKey(r.head_sha)
@@ -331,7 +346,7 @@ export function createAiGatewayGraphContract(kit) {
     {
       kind: 'edge',
       type: 'in',
-      sql: `SELECT head_sha, git_remote, message_created_at FROM ${SOURCE_DATASET}`,
+      columns: ['head_sha', 'git_remote', 'message_created_at'],
       toRow(r) {
         const commit = keys.commitKey(r.head_sha)
         const repo = keys.repoKeyFromRemote(r.git_remote)
@@ -349,7 +364,8 @@ export function createAiGatewayGraphContract(kit) {
     {
       kind: 'edge',
       type: 'invoked',
-      sql: `SELECT session_id, tool_name, tool_args, message_created_at FROM ${SOURCE_DATASET} WHERE part_type = 'tool_call' AND tool_name IN ('Bash', 'exec_command')`,
+      columns: ['session_id', 'tool_name', 'tool_args', 'message_created_at'],
+      where: { eq: { part_type: 'tool_call' }, in: { tool_name: ['Bash', 'exec_command'] } },
       toRow(r) {
         const session = str(r.session_id)
         const program = programFrom(commandStringFrom(r.tool_name, r.tool_args))
@@ -373,7 +389,8 @@ export function createAiGatewayGraphContract(kit) {
     {
       kind: 'edge',
       type: 'ran',
-      sql: `SELECT session_id, tool_args, message_created_at FROM ${SOURCE_DATASET} WHERE part_type = 'tool_call' AND tool_name = 'Skill'`,
+      columns: ['session_id', 'tool_args', 'message_created_at'],
+      where: { eq: { part_type: 'tool_call', tool_name: 'Skill' } },
       toRow(r) {
         const session = str(r.session_id)
         const skill = skillFromToolArgs(r.tool_args)
@@ -388,7 +405,12 @@ export function createAiGatewayGraphContract(kit) {
     {
       kind: 'edge',
       type: 'ran',
-      sql: `SELECT session_id, content_text, message_created_at FROM ${SOURCE_DATASET} WHERE role = 'user' AND part_type = 'text' AND content_text LIKE 'Base directory for this skill: %'`,
+      // Raw SQL on purpose: the prefix filter prunes `content_text` (the
+      // table's largest column) server-side, keeping it out of the shared
+      // scan's materialization. `attributes` rides for the contract
+      // rowFilter (registry-enforced).
+      // @ref LLP 0096#decision [constrained-by]: heavy-column prefix guards stay pushed down
+      sql: `SELECT attributes, session_id, content_text, message_created_at FROM ${SOURCE_DATASET} WHERE role = 'user' AND part_type = 'text' AND content_text LIKE 'Base directory for this skill: %'`,
       toRow(r) {
         const session = str(r.session_id)
         const skill = skillFromMarker(r.content_text)
@@ -401,7 +423,8 @@ export function createAiGatewayGraphContract(kit) {
     {
       kind: 'edge',
       type: 'ran',
-      sql: `SELECT session_id, content_text, message_created_at FROM ${SOURCE_DATASET} WHERE role = 'user' AND part_type = 'text' AND content_text LIKE '<command-name>%'`,
+      // Raw SQL on purpose, same pushdown rationale as the marker surface.
+      sql: `SELECT attributes, session_id, content_text, message_created_at FROM ${SOURCE_DATASET} WHERE role = 'user' AND part_type = 'text' AND content_text LIKE '<command-name>%'`,
       toRow(r) {
         const session = str(r.session_id)
         const skill = skillFromSlash(r.content_text)
@@ -417,7 +440,8 @@ export function createAiGatewayGraphContract(kit) {
     {
       kind: 'edge',
       type: 'ran',
-      sql: `SELECT session_id, tool_args, message_created_at FROM ${SOURCE_DATASET} WHERE part_type = 'tool_call' AND tool_name = 'exec_command'`,
+      columns: ['session_id', 'tool_args', 'message_created_at'],
+      where: { eq: { part_type: 'tool_call', tool_name: 'exec_command' } },
       toRow(r) {
         const session = str(r.session_id)
         const skill = skillFromCodexRead(commandStringFrom('exec_command', r.tool_args))
@@ -427,45 +451,32 @@ export function createAiGatewayGraphContract(kit) {
     },
   ]
 
-  // @ref LLP 0026#decision [implements]: tag-don't-drop: the gateway now
-  // RETAINS Claude harness aux exchanges (security monitor, etc.) tagged
-  // `attributes.claude.aux_kind` instead of dropping them. That traffic is
-  // real but not user conversation, so it must not mint graph
-  // Session/App/Model/Tool/File nodes or edges. One shared source filter:
-  // select `attributes` for every rule and drop aux rows before toRow runs,
-  // so each rule's SQL stays focused on its own columns.
-  const auxFilteredRules = rules.map((rule) => ({
-    ...rule,
-    sql: withAttributes(rule.sql),
-    /** @param {Record<string, unknown>} r */
-    toRow(r) {
-      if (auxKindOf(r.attributes)) return null
-      return rule.toRow(r)
-    },
-  }))
-
   return {
     name: 'ai-gateway-t0',
     plugin: PLUGIN_NAME,
     sourceDataset: SOURCE_DATASET,
     projector: PROJECTOR,
     projectorVersion: PROJECTOR_VERSION,
-    rules: auxFilteredRules,
+    rules,
+    // @ref LLP 0026#decision [implements]: tag-don't-drop: the gateway
+    // RETAINS Claude harness aux exchanges (security monitor, etc.) tagged
+    // `attributes.claude.aux_kind` instead of dropping them. That traffic is
+    // real but not user conversation, so it must not mint graph
+    // Session/App/Model/Tool/File nodes or edges.
+    // @ref LLP 0096#decision [implements]: the aux filter is a contract
+    // rowFilter, evaluated once per source row by the engine (both paths),
+    // instead of the old per-rule SQL prepend + toRow wrap that parsed
+    // `attributes` once per rule per row.
+    rowFilter: {
+      columns: ['attributes'],
+      /** @param {Record<string, unknown>} r */
+      keep(r) {
+        return auxKindOf(r.attributes) === null
+      },
+    },
   }
 }
 
-/**
- * Prepend the `attributes` column to a rule's projection so the shared aux
- * filter can read `attributes.claude.aux_kind` without each rule's SQL
- * repeating it. Every rule begins `SELECT <cols> FROM …`.
- *
- * @param {string} sql
- * @returns {string}
- */
-function withAttributes(sql) {
-  return sql.replace(/^SELECT\s+/i, 'SELECT attributes, ')
-}
-
 /**
  * Read `attributes.claude.aux_kind` from a source row. `attributes` is a
  * JSON column that may arrive parsed or as a string (like `tool_args`).
diff --git a/hypaware-core/plugins-workspace/ai-gateway-graph/src/types.d.ts b/hypaware-core/plugins-workspace/ai-gateway-graph/src/types.d.ts
index a449fa8..9996ec5 100644
--- a/hypaware-core/plugins-workspace/ai-gateway-graph/src/types.d.ts
+++ b/hypaware-core/plugins-workspace/ai-gateway-graph/src/types.d.ts
@@ -3,11 +3,24 @@
 /** A materialized graph row (node or edge), keyed by column name. */
 export type GraphRow = Record<string, unknown>
 
-/** One T0 contract rule: a read-only SELECT plus a row mapper. */
+/** A declarative rule filter (LLP 0096): AND of eq / in / likePrefix, SQL null semantics. */
+export interface RulePredicate {
+  eq?: Record<string, string>
+  in?: Record<string, string[]>
+  likePrefix?: Record<string, string>
+}
+
+/**
+ * One T0 contract rule: a source read plus a row mapper. Declarative
+ * `columns` (+ optional `where`) joins the contract's shared scan; raw `sql`
+ * runs standalone (LLP 0096). Exactly one of the two.
+ */
 export interface ContractRule {
   kind: 'node' | 'edge'
   type: string
-  sql: string
+  sql?: string
+  columns?: string[]
+  where?: RulePredicate
   toRow(row: Record<string, unknown>): GraphRow | null
 }
 
diff --git a/hypaware-core/plugins-workspace/context-graph/src/contract-registry.js b/hypaware-core/plugins-workspace/context-graph/src/contract-registry.js
index 1c67b79..9bbff5c 100644
--- a/hypaware-core/plugins-workspace/context-graph/src/contract-registry.js
+++ b/hypaware-core/plugins-workspace/context-graph/src/contract-registry.js
@@ -48,10 +48,24 @@ export function createContractRegistry(opts = {}) {
       throw new TypeError(`registerContract: '${contract.name}' rules must be a non-empty array`)
     }
     // Validate each rule's shape at registration, not at projection time: the
-    // engine reads `kind`/`sql`/`toRow` directly (project.js) and routes by
-    // `kind`, so a connector typo would otherwise surface as a confusing
-    // mid-projection failure (or silently route rows into the wrong target
-    // map) far from the contract that caused it.
+    // engine reads `kind`/`sql`/`columns`/`where`/`toRow` directly
+    // (project.js) and routes by `kind`, so a connector typo would otherwise
+    // surface as a confusing mid-projection failure (or silently route rows
+    // into the wrong target map) far from the contract that caused it.
+    // @ref LLP 0096#decision [implements]: exactly one read form per rule; `where` only rides `columns`; raw SQL must carry the rowFilter's columns itself
+    if (contract.rowFilter !== undefined) {
+      const filter = contract.rowFilter
+      const at = `'${contract.name}' rowFilter`
+      if (!filter || typeof filter !== 'object') {
+        throw new TypeError(`registerContract: ${at} must be an object`)
+      }
+      if (!Array.isArray(filter.columns) || filter.columns.length === 0 || filter.columns.some((c) => typeof c !== 'string' || c.length === 0)) {
+        throw new TypeError(`registerContract: ${at} columns must be non-empty strings`)
+      }
+      if (typeof filter.keep !== 'function') {
+        throw new TypeError(`registerContract: ${at} keep must be a function`)
+      }
+    }
     contract.rules.forEach((rule, i) => {
       const at = `'${contract.name}' rule ${i}`
       if (!rule || typeof rule !== 'object') {
@@ -63,8 +77,27 @@ export function createContractRegistry(opts = {}) {
       if (typeof rule.type !== 'string' || rule.type.length === 0) {
         throw new TypeError(`registerContract: ${at} type must be a non-empty string`)
       }
-      if (typeof rule.sql !== 'string' || rule.sql.length === 0) {
-        throw new TypeError(`registerContract: ${at} sql must be a non-empty string`)
+      const hasSql = typeof rule.sql === 'string' && rule.sql.length > 0
+      const hasColumns = Array.isArray(rule.columns)
+      if (hasSql === hasColumns) {
+        throw new TypeError(`registerContract: ${at} must carry exactly one of sql or columns`)
+      }
+      if (hasColumns) {
+        const cols = /** @type {unknown[]} */ (rule.columns)
+        if (cols.length === 0 || cols.some((c) => typeof c !== 'string' || c.length === 0)) {
+          throw new TypeError(`registerContract: ${at} columns must be non-empty strings`)
+        }
+        if (rule.where !== undefined) validatePredicate(rule.where, at)
+      } else if (rule.where !== undefined) {
+        throw new TypeError(`registerContract: ${at} where is only valid with columns`)
+      }
+      if (hasSql && contract.rowFilter) {
+        const sql = /** @type {string} */ (rule.sql)
+        for (const col of contract.rowFilter.columns) {
+          if (!rawSqlProjectsColumn(sql, col)) {
+            throw new TypeError(`registerContract: ${at} raw sql must select rowFilter column '${col}'`)
+          }
+        }
       }
       if (typeof rule.toRow !== 'function') {
         throw new TypeError(`registerContract: ${at} toRow must be a function`)
@@ -86,6 +119,48 @@ export function createContractRegistry(opts = {}) {
     })
   }
 
+  /**
+   * A `where` must be built from the three supported predicate shapes only,
+   * with the value types the JS evaluator expects: anything else would
+   * silently match nothing at projection time.
+   *
+   * @param {unknown} where
+   * @param {string} at
+   */
+  function validatePredicate(where, at) {
+    if (!where || typeof where !== 'object') {
+      throw new TypeError(`registerContract: ${at} where must be an object`)
+    }
+    const w = /** @type {Record<string, unknown>} */ (where)
+    for (const key of Object.keys(w)) {
+      if (key !== 'eq' && key !== 'in' && key !== 'likePrefix') {
+        throw new TypeError(`registerContract: ${at} where.${key} is not a supported predicate (eq, in, likePrefix)`)
+      }
+    }
+    for (const shape of ['eq', 'likePrefix']) {
+      const block = w[shape]
+      if (block === undefined) continue
+      if (!block || typeof block !== 'object') {
+        throw new TypeError(`registerContract: ${at} where.${shape} must be an object`)
+      }
+      for (const [col, value] of Object.entries(block)) {
+        if (typeof value !== 'string' || value.length === 0) {
+          throw new TypeError(`registerContract: ${at} where.${shape}.${col} must be a non-empty string`)
+        }
+      }
+    }
+    if (w.in !== undefined) {
+      if (!w.in || typeof w.in !== 'object') {
+        throw new TypeError(`registerContract: ${at} where.in must be an object`)
+      }
+      for (const [col, list] of Object.entries(w.in)) {
+        if (!Array.isArray(list) || list.length === 0 || list.some((v) => typeof v !== 'string' || v.length === 0)) {
+          throw new TypeError(`registerContract: ${at} where.in.${col} must be a non-empty array of strings`)
+        }
+      }
+    }
+  }
+
   /**
    * All registered contracts, name-sorted so projection order is stable.
    * @returns {Contract[]}
@@ -96,3 +171,127 @@ export function createContractRegistry(opts = {}) {
 
   return { register, list }
 }
+
+/**
+ * True when a raw rule's SQL provably projects `col` in its top-level
+ * SELECT list, so the contract's rowFilter (which reads `row[col]`) has the
+ * column to test. A loose `sql.includes(col)` accepts false positives (a
+ * `WHERE attributes IS NOT NULL`, or a different column whose name merely
+ * contains `col`), so match the projection list only: take the identifiers
+ * between the first top-level SELECT and its FROM, accept `*` / `table.*`,
+ * and reduce each item to its output name (the alias after AS, or the column
+ * past a `table.` qualifier). Anything ambiguous (a computed expression with
+ * no alias) is treated as not-a-match, so the guard stays conservative and
+ * rejects registration when the column is not provably projected. This is a
+ * focused projection check, not a general SQL parser.
+ *
+ * @param {string} sql
+ * @param {string} col
+ * @returns {boolean}
+ */
+function rawSqlProjectsColumn(sql, col) {
+  const projection = selectProjection(sql)
+  if (projection === undefined) return false
+  for (const item of splitTopLevel(projection)) {
+    const name = projectionOutputName(item)
+    if (name === '*' || name === col) return true
+  }
+  return false
+}
+
+/**
+ * The text between the first top-level `SELECT` and its matching `FROM`
+ * (both matched as standalone, case-insensitive keywords at parenthesis
+ * depth 0, so a subquery's SELECT/FROM never leaks in). Undefined when the
+ * SQL has no top-level `SELECT ... FROM`.
+ *
+ * @param {string} sql
+ * @returns {string | undefined}
+ */
+function selectProjection(sql) {
+  const upper = sql.toUpperCase()
+  let depth = 0
+  let selectEnd = -1
+  for (let i = 0; i < sql.length; i++) {
+    const ch = sql[i]
+    if (ch === '(') depth++
+    else if (ch === ')') depth--
+    else if (depth === 0 && selectEnd === -1 && matchKeyword(upper, i, 'SELECT')) {
+      selectEnd = i + 'SELECT'.length
+      i = selectEnd - 1
+    } else if (depth === 0 && selectEnd !== -1 && matchKeyword(upper, i, 'FROM')) {
+      return sql.slice(selectEnd, i)
+    }
+  }
+  return undefined
+}
+
+/**
+ * True when `kw` sits at index `i` of `upper` as a whole word (its
+ * neighbours are not identifier characters), so `FROM` matches but
+ * `FROMAGE` or a `from_x` column does not.
+ *
+ * @param {string} upper
+ * @param {number} i
+ * @param {string} kw
+ * @returns {boolean}
+ */
+function matchKeyword(upper, i, kw) {
+  if (!upper.startsWith(kw, i)) return false
+  const boundary = (/** @type {string | undefined} */ c) => c === undefined || !/[A-Z0-9_]/.test(c)
+  return boundary(upper[i - 1]) && boundary(upper[i + kw.length])
+}
+
+/**
+ * Split on commas at parenthesis depth 0, so a `f(a, b)` projection item
+ * stays whole.
+ *
+ * @param {string} s
+ * @returns {string[]}
+ */
+function splitTopLevel(s) {
+  /** @type {string[]} */
+  const parts = []
+  let depth = 0
+  let start = 0
+  for (let i = 0; i < s.length; i++) {
+    const ch = s[i]
+    if (ch === '(') depth++
+    else if (ch === ')') depth--
+    else if (ch === ',' && depth === 0) {
+      parts.push(s.slice(start, i))
+      start = i + 1
+    }
+  }
+  parts.push(s.slice(start))
+  return parts
+}
+
+/**
+ * The output name a single projection item exposes on the result row: the
+ * alias after `AS`, `*` for a wildcard (`*` or `table.*`), or the column
+ * past a `table.` qualifier. A bare computed expression has no derivable
+ * column name and returns its trimmed text, which will not match a plain
+ * column name, keeping the guard conservative.
+ *
+ * @param {string} item
+ * @returns {string}
+ */
+function projectionOutputName(item) {
+  let s = item.trim()
+  if (s.length === 0) return ''
+  const asMatch = /\s+AS\s+("?[A-Za-z0-9_]+"?)\s*$/i.exec(s)
+  if (asMatch) return stripQuotes(asMatch[1])
+  if (s === '*' || s.endsWith('.*')) return '*'
+  const dot = s.lastIndexOf('.')
+  if (dot !== -1) s = s.slice(dot + 1)
+  return stripQuotes(s)
+}
+
+/**
+ * @param {string} s
+ * @returns {string}
+ */
+function stripQuotes(s) {
+  return s.startsWith('"') && s.endsWith('"') && s.length >= 2 ? s.slice(1, -1) : s
+}
diff --git a/hypaware-core/plugins-workspace/context-graph/src/project.js b/hypaware-core/plugins-workspace/context-graph/src/project.js
index b5a5758..4c2e140 100644
--- a/hypaware-core/plugins-workspace/context-graph/src/project.js
+++ b/hypaware-core/plugins-workspace/context-graph/src/project.js
@@ -14,7 +14,7 @@ import {
 /**
  * @import { HypAwareV2Config, QueryRegistry } from '../../../../hypaware-plugin-kernel-types.js'
  * @import { ExtendedQueryStorageService } from '../../../../src/core/cache/types.js'
- * @import { Contract, GraphRow } from './types.js'
+ * @import { Contract, ContractRule, GraphRow, RulePredicate } from './types.js'
  */
 
 /**
@@ -45,26 +45,89 @@ export async function projectGraph({ query, storage, contracts, config, dryRun =
       /** @type {Map<string, GraphRow>} */
       const edges = new Map()
 
+      /**
+       * Feed one source row through a set of rules into the node/edge maps.
+       * Merge is order-independent (`mergeRow`), so fanning a shared scan's
+       * row out to many rules is observationally identical to the old
+       * one-query-per-rule order.
+       *
+       * @param {Record<string, unknown>} row
+       * @param {ContractRule[]} rules
+       */
+      const applyRules = (row, rules) => {
+        for (const rule of rules) {
+          if (rule.where && !matchesPredicate(rule.where, row)) continue
+          const built = rule.toRow(row)
+          if (!built) continue
+          const target = rule.kind === 'node' ? nodes : edges
+          const idKey = rule.kind === 'node' ? 'node_id' : 'edge_id'
+          const id = /** @type {string} */ (built[idKey])
+          const existing = target.get(id)
+          if (existing) mergeRow(existing, built)
+          else target.set(id, built)
+        }
+      }
+
       let sourceRows = 0
+      let scanCount = 0
       for (const contract of contracts) {
-        for (const rule of contract.rules) {
+        const keep = contract.rowFilter?.keep
+        const declarative = contract.rules.filter((rule) => rule.columns)
+        const raw = contract.rules.filter((rule) => typeof rule.sql === 'string')
+
+        // One shared scan per contract: the union of the declarative rules'
+        // columns (plus the row filter's), each rule's predicate evaluated
+        // in JS per row. This is the whole LLP 0095 fix: rule count no
+        // longer multiplies table scans, and the contract row filter runs
+        // once per row instead of once per rule per row.
+        // @ref LLP 0096#decision [implements]: one scan per contract; JS predicates with SQL null semantics; rowFilter once per row
+        if (declarative.length > 0) {
+          const columns = new Set()
+          for (const rule of declarative) {
+            for (const col of rule.columns ?? []) columns.add(col)
+            for (const col of predicateColumns(rule.where)) columns.add(col)
+          }
+          for (const col of contract.rowFilter?.columns ?? []) columns.add(col)
           const result = await executeQuerySql({
-            query: rule.sql,
+            query: `SELECT ${[...columns].sort().join(', ')} FROM ${contract.sourceDataset}`,
             registry: query,
             storage,
             config,
             refresh: 'always',
           })
           sourceRows += result.rows.length
-          const target = rule.kind === 'node' ? nodes : edges
-          const idKey = rule.kind === 'node' ? 'node_id' : 'edge_id'
+          scanCount += 1
+          for (const row of result.rows) {
+            if (keep && !keep(row)) continue
+            applyRules(row, declarative)
+          }
+        }
+
+        // Raw-SQL rules run standalone, grouped by identical SQL text so
+        // rule pairs sharing a query (a surface's node + edge rule) cost one
+        // scan. Their SQL already selects the rowFilter's columns (registry
+        // enforced), so the filter applies here too.
+        /** @type {Map<string, ContractRule[]>} */
+        const bySql = new Map()
+        for (const rule of raw) {
+          const sql = /** @type {string} */ (rule.sql)
+          const group = bySql.get(sql)
+          if (group) group.push(rule)
+          else bySql.set(sql, [rule])
+        }
+        for (const [sql, rules] of bySql) {
+          const result = await executeQuerySql({
+            query: sql,
+            registry: query,
+            storage,
+            config,
+            refresh: 'always',
+          })
+          sourceRows += result.rows.length
+          scanCount += 1
           for (const row of result.rows) {
-            const built = rule.toRow(row)
-            if (!built) continue
-            const id = /** @type {string} */ (built[idKey])
-            const existing = target.get(id)
-            if (existing) mergeRow(existing, built)
-            else target.set(id, built)
+            if (keep && !keep(row)) continue
+            applyRules(row, rules)
           }
         }
       }
@@ -72,6 +135,7 @@ export async function projectGraph({ query, storage, contracts, config, dryRun =
       const nodeRows = [...nodes.values()]
       const edgeRows = [...edges.values()]
       span.setAttribute('contract_count', contracts.length)
+      span.setAttribute('scan_count', scanCount)
       span.setAttribute('source_row_count', sourceRows)
       span.setAttribute('node_count', nodeRows.length)
       span.setAttribute('edge_count', edgeRows.length)
@@ -144,6 +208,54 @@ async function dedupExisting(rows, idCol, dataset, query, storage, config) {
   return rows.filter((r) => !seen.has(/** @type {string} */ (r[idCol])))
 }
 
+/**
+ * Evaluate a rule's declarative predicate against one source row with SQL
+ * semantics: every clause must hold, and a null or absent column never
+ * matches (`WHERE col = 'x'` skips null rows; so does this).
+ *
+ * @param {RulePredicate} where
+ * @param {Record<string, unknown>} row
+ * @returns {boolean}
+ * @ref LLP 0096#decision [implements]: eq/in/likePrefix only, AND-composed, null never matches
+ */
+export function matchesPredicate(where, row) {
+  if (where.eq) {
+    for (const [col, value] of Object.entries(where.eq)) {
+      if (row[col] !== value) return false
+    }
+  }
+  if (where.in) {
+    for (const [col, values] of Object.entries(where.in)) {
+      const v = row[col]
+      if (typeof v !== 'string' || !values.includes(v)) return false
+    }
+  }
+  if (where.likePrefix) {
+    for (const [col, prefix] of Object.entries(where.likePrefix)) {
+      const v = row[col]
+      if (typeof v !== 'string' || !v.startsWith(prefix)) return false
+    }
+  }
+  return true
+}
+
+/**
+ * The columns a predicate reads, so the shared scan selects them even when
+ * no rule's `columns` lists them (a rule may filter on a column `toRow`
+ * never touches).
+ *
+ * @param {RulePredicate | undefined} where
+ * @returns {string[]}
+ */
+function predicateColumns(where) {
+  if (!where) return []
+  return [
+    ...Object.keys(where.eq ?? {}),
+    ...Object.keys(where.in ?? {}),
+    ...Object.keys(where.likePrefix ?? {}),
+  ]
+}
+
 /**
  * True when a dedup query failed because the dataset is not queryable
  * yet (unregistered, or its backing path is absent) rather than because
diff --git a/hypaware-core/plugins-workspace/context-graph/src/types.d.ts b/hypaware-core/plugins-workspace/context-graph/src/types.d.ts
index b1deb2a..6c2e630 100644
--- a/hypaware-core/plugins-workspace/context-graph/src/types.d.ts
+++ b/hypaware-core/plugins-workspace/context-graph/src/types.d.ts
@@ -16,11 +16,37 @@ export interface SkippedPartition {
   reason: 'unreadable-cursor' | 'unexpected-layout' | 'concurrent-write'
 }
 
-/** One T0 contract rule: a read-only SELECT plus a row mapper. */
+/**
+ * A declarative rule filter: the conjunction of the three predicate shapes
+ * contract rules actually use, evaluated in JS against the contract's one
+ * shared scan with SQL semantics (a null or absent column never matches).
+ * No general SQL parsing, no expression language (LLP 0096 §1).
+ */
+export interface RulePredicate {
+  /** column = value */
+  eq?: Record<string, string>
+  /** column IN (values) */
+  in?: Record<string, string[]>
+  /** column LIKE 'prefix%' */
+  likePrefix?: Record<string, string>
+}
+
+/**
+ * One T0 contract rule: a source read plus a row mapper. Exactly one read
+ * form: declarative `columns` (+ optional `where`) joins the contract's one
+ * shared scan; raw `sql` runs standalone, for contracts that migrate on
+ * their own schedule or predicates that must prune a heavy column
+ * server-side (LLP 0096 §2-3).
+ */
 export interface ContractRule {
   kind: 'node' | 'edge'
   type: string
-  sql: string
+  /** Raw SELECT escape hatch; mutually exclusive with `columns`. Must select the contract rowFilter's columns itself. */
+  sql?: string
+  /** Columns `toRow` reads; the shared scan selects the union across rules. Mutually exclusive with `sql`. */
+  columns?: string[]
+  /** Declarative filter; only valid with `columns`. */
+  where?: RulePredicate
   toRow(row: Record<string, unknown>): GraphRow | null
 }
 
@@ -82,6 +108,16 @@ export interface Contract {
   projectorVersion: number
   /** The node/edge rules the engine runs for this source. */
   rules: ContractRule[]
+  /**
+   * Optional source-row filter evaluated once per row before any rule's
+   * `toRow`, on both the shared-scan and raw-SQL paths (LLP 0096 §4). The
+   * engine adds `columns` to the shared scan; raw-SQL rules must select
+   * them in their own SQL (the registry enforces this textually).
+   */
+  rowFilter?: {
+    columns: string[]
+    keep(row: Record<string, unknown>): boolean
+  }
 }
 
 /** The in-plugin registry source plugins contribute contracts into. */
diff --git a/llp/0095-projection-read-amplification.issue.md b/llp/0095-projection-read-amplification.issue.md
new file mode 100644
index 0000000..ca04d41
--- /dev/null
+++ b/llp/0095-projection-read-amplification.issue.md
@@ -0,0 +1,66 @@
+# LLP 0095: graph projection reads the source table once per rule
+
+**Type:** Issue
+**Status:** Active
+**Systems:** Plugins
+**Author:** Phil / Claude
+**Date:** 2026-07-09
+**Related:** LLP 0023, LLP 0026, LLP 0073, LLP 0077
+
+> `projectGraph` executes every contract rule as an independent full SQL scan
+> of the source dataset. With 25 rules on the ai-gateway contract, one
+> projection reads the source 25 times and JSON-parses the `attributes`
+> column 25 times per row. Invisible on a small local cache; pathological at
+> server scale, and fatal to automatic projection (PR #279) at any scale.
+
+## Observed
+
+On a production hypaware-server deployment:
+
+- Source: `ai_gateway_messages`, ~700k rows, ~1.3GB live cache across ~1,300
+  parquet files.
+- A single org-scoped `graph project` dry run read **~35GB** from disk and
+  had not finished after **45 minutes**, single core pegged, before it was
+  aborted. 25 rules x 1.3GB accounts for ~33GB of that.
+- The same corpus projects in seconds on a laptop (about 4x fewer rows,
+  3x fewer files, faster storage). The cost curve is
+  `rules x table-size x file-count`; local installs are on the same curve
+  and merely earlier on it.
+
+## Why
+
+Three stacked multipliers in `context-graph/src/project.js` and the contract
+shape it consumes:
+
+1. **Per-rule scans.** The engine loops `for contract / for rule` and calls
+   `executeQuerySql(rule.sql)` per rule. Every rule of the ai-gateway
+   contract selects from the same table; nothing is shared between the 25
+   queries (LLP 0023 settled the contract seam, not this execution shape).
+2. **The aux filter rides every rule.** LLP 0026's tag-don't-drop filter
+   wraps each rule: `withAttributes` prepends the `attributes` column
+   (19% of table bytes) to every rule's SQL, and `auxKindOf` JSON-parses it
+   again for every rule that sees the row. One row's `attributes` is parsed
+   up to 25 times per projection.
+3. **`refresh: 'always'` per rule.** Each rule query re-runs the partition
+   refresh pass, multiplying per-partition overhead by rule count.
+
+## Impact
+
+- Server-side projection (hypaware-server LLP 0037) is effectively unusable
+  on real orgs: the admin endpoint times out and burns ~2 cores for the
+  better part of an hour per run.
+- PR #279 (LLP 0086-0092, automatic projection on a daemon tick, Draft)
+  assumes projection is cheap enough to run every 15 minutes. This issue is
+  a blocker for that design's viability at any accumulated history.
+- Adding a rule to a contract today costs one more full table scan per
+  projection, which mis-prices what should be a cheap contribution
+  (LLP 0023 §contract-contribution).
+
+## Fix
+
+Settled in [LLP 0096](./0096-shared-scan-projection.decision.md): one shared
+scan per contract with declarative per-rule predicates evaluated in JS, the
+aux filter evaluated once per row, raw-SQL rules retained as a standalone
+path. Incremental (watermarked) projection is out of scope here; it belongs
+to the automatic-derivation track (PR #279) as steady-state work atop this
+fix.
diff --git a/llp/0096-shared-scan-projection.decision.md b/llp/0096-shared-scan-projection.decision.md
new file mode 100644
index 0000000..a3f7b5d
--- /dev/null
+++ b/llp/0096-shared-scan-projection.decision.md
@@ -0,0 +1,66 @@
+# LLP 0096: one shared scan per contract, declarative rule predicates
+
+**Type:** Decision
+**Status:** Accepted
+**Systems:** Plugins
+**Author:** Phil / Claude
+**Date:** 2026-07-09
+**Related:** LLP 0095, LLP 0023, LLP 0026
+
+> Fixes [LLP 0095](./0095-projection-read-amplification.issue.md) by changing
+> how the projection engine executes a contract, not what a contract means:
+> rules declare columns and a simple predicate instead of raw SQL, the engine
+> scans the source once per contract, and the LLP 0026 aux filter becomes a
+> contract-level row filter evaluated once per row.
+
+## Decision
+
+1. **`ContractRule` gains a declarative form.** A rule carries either
+   `{ columns, where? }` or the legacy `{ sql }`, never both. `where` is a
+   conjunction of the three predicate shapes the existing rules actually use:
+   `eq` (column = value), `in` (column in list), `likePrefix`
+   (column LIKE 'prefix%'). No general SQL parsing, no expression language.
+2. **One scan per contract for declarative rules.** The engine unions the
+   declarative rules' `columns` (plus the contract row filter's columns),
+   runs a single `SELECT <union> FROM <sourceDataset>` per contract, and
+   evaluates each rule's `where` in JS per row before `toRow`. Predicate
+   semantics mirror SQL: a null or absent column never matches.
+3. **Raw-SQL rules stay supported and run standalone**, grouped by identical
+   SQL text so duplicate queries collapse. This keeps two escape hatches
+   open: contracts in other repos migrate on their own schedule (the
+   server's github contract, context-graph-enrich), and rules whose
+   predicate must prune a heavy column server-side stay pushed down (the
+   Skill surface rules' `content_text LIKE 'prefix%'` guards, which keep the
+   largest column out of the shared scan's materialization).
+4. **The aux filter moves to the contract.** `Contract` gains optional
+   `rowFilter: { columns, keep(row) }`, evaluated once per row on both the
+   shared scan and raw-SQL paths. The ai-gateway contract's per-rule
+   `withAttributes`/`auxKindOf` wrapping (LLP 0026) is replaced by one
+   `rowFilter` that parses `attributes` once per row instead of once per
+   rule per row.
+5. **`refresh: 'always'` runs once per distinct query**, which the shared
+   scan reduces to approximately once per contract.
+
+## Alternatives rejected
+
+- **Temp slim-table**: materialize the union columns into a temp table and
+  point the untouched rule SQL at it. Rejected: introduces temp-dataset
+  plumbing with no other consumer, and still pays 25 query executions.
+- **Group rules by identical WHERE text only**: no contract-shape change,
+  but the ai-gateway contract still needs ~7 scans and the per-rule
+  `attributes` parsing remains. Insufficient for LLP 0095's numbers.
+- **Incremental (watermarked) projection**: complementary, not competing;
+  belongs to the automatic-derivation track (PR #279). This decision makes
+  every run cheap; a watermark would additionally shrink what a run reads.
+
+## Consequences
+
+- The registry validates the new shape (exactly one of `sql` or `columns`);
+  `kind`/`type`/`toRow` validation is unchanged.
+- Projection output is bit-identical: same rows reach the same `toRow`
+  functions in the same contract order; only the read path changes. The e2e
+  regression (test/plugins/context-graph-project-e2e.test.js) plus a
+  shared-vs-sql equivalence test hold that line.
+- Expected effect at LLP 0095's prod scale: 25 full scans to 1 slim scan
+  plus 2 pushed-down prefix scans, and 25x fewer `attributes` parses;
+  minutes, not the better part of an hour.
diff --git a/test/plugins/ai-gateway-graph-contract.test.js b/test/plugins/ai-gateway-graph-contract.test.js
index 989b205..5a8571f 100644
--- a/test/plugins/ai-gateway-graph-contract.test.js
+++ b/test/plugins/ai-gateway-graph-contract.test.js
@@ -175,35 +175,24 @@ test('numeric natural keys are stringified', () => {
 // @ref LLP 0026#decision: Claude harness aux exchanges are retained (tagged
 // attributes.claude.aux_kind) rather than dropped; the graph contract must
 // exclude them so security-monitor traffic mints no Session/App/Tool noise.
-test('every rule selects attributes so the shared aux filter has its input', () => {
-  for (const r of contract.rules) {
-    assert.match(r.sql, /^SELECT\s+attributes\b/i, `rule ${r.kind}/${r.type} projects attributes`)
+// @ref LLP 0096#decision [tests]: the aux filter is the contract rowFilter,
+// evaluated once per source row by the engine (both scan paths), so these
+// assertions target keep() rather than each rule's toRow. The end-to-end
+// exclusion (rows never reaching any rule) is pinned by the projection e2e.
+test('the contract declares the aux rowFilter on attributes, and raw rules select it', () => {
+  assert.ok(contract.rowFilter, 'contract carries the aux rowFilter')
+  assert.deepEqual(contract.rowFilter.columns, ['attributes'])
+  for (const r of contract.rules.filter((r) => typeof r.sql === 'string')) {
+    assert.match(/** @type {string} */ (r.sql), /^SELECT\s+attributes\b/i, `raw rule ${r.kind}/${r.type} projects attributes itself`)
   }
 })
 
-test('aux-tagged rows are excluded from every node and edge rule', () => {
-  const aux = { attributes: { claude: { aux_kind: 'security_monitor' } } }
-  // Otherwise-valid rows that would each mint a node/edge if not aux.
-  assert.equal(rule('node', 'Session').toRow({ ...aux, conversation_id: 'conv-1', message_created_at: TS }), null)
-  assert.equal(rule('node', 'App').toRow({ ...aux, client_name: 'claude', message_created_at: TS }), null)
-  assert.equal(rule('node', 'Model').toRow({ ...aux, model: 'claude-opus', message_created_at: TS }), null)
-  assert.equal(rule('node', 'Tool').toRow({ ...aux, tool_name: 'Read', part_type: 'tool_call', message_created_at: TS }), null)
-  assert.equal(rule('node', 'File').toRow({ ...aux, tool_name: 'Read', tool_args: { file_path: '/x' }, message_created_at: TS }), null)
-  assert.equal(rule('edge', 'via').toRow({ ...aux, conversation_id: 'c', client_name: 'a', message_created_at: TS }), null)
-  assert.equal(rule('edge', 'used_model').toRow({ ...aux, conversation_id: 'c', model: 'm', message_created_at: TS }), null)
-  assert.equal(rule('edge', 'used').toRow({ ...aux, conversation_id: 'c', tool_name: 'Read', message_created_at: TS }), null)
-  assert.equal(rule('edge', 'touched').toRow({ ...aux, conversation_id: 'c', tool_name: 'Write', tool_args: { file_path: '/a.js' }, message_created_at: TS }), null)
-  assert.equal(rule('node', 'Program').toRow({ ...aux, tool_name: 'Bash', tool_args: { command: 'git status' }, part_type: 'tool_call', message_created_at: TS }), null)
-  assert.equal(rule('edge', 'invoked').toRow({ ...aux, session_id: 's', tool_name: 'Bash', tool_args: { command: 'git status' }, part_type: 'tool_call', message_created_at: TS }), null)
-})
-
-test('aux filter handles attributes arriving as a JSON string', () => {
-  const row = rule('node', 'Session').toRow({
-    attributes: JSON.stringify({ claude: { aux_kind: 'security_monitor' } }),
-    conversation_id: 'conv-1',
-    message_created_at: TS,
-  })
-  assert.equal(row, null, 'aux_kind in a stringified attributes column is still excluded')
+test('the aux rowFilter drops aux-tagged rows and keeps real ones', () => {
+  const { keep } = /** @type {NonNullable<typeof contract.rowFilter>} */ (contract.rowFilter)
+  assert.equal(keep({ attributes: { claude: { aux_kind: 'security_monitor' } } }), false, 'aux object excluded')
+  assert.equal(keep({ attributes: JSON.stringify({ claude: { aux_kind: 'security_monitor' } }) }), false, 'aux_kind in a stringified attributes column is still excluded')
+  assert.equal(keep({ attributes: { claude: { client_name: 'claude' }, dev_run_id: 'run-1' } }), true, 'attributes without aux_kind pass')
+  assert.equal(keep({ attributes: null }), true, 'absent attributes pass')
 })
 
 test('non-aux rows pass through unchanged (attributes present but no aux_kind)', () => {
@@ -356,8 +345,8 @@ test('Commit -in-> Repo is the second `in` edge and converges with the GitHub ed
 
 test('Program/invoked rules select only tool_call rows from the two shell tools', () => {
   for (const r of [rule('node', 'Program'), rule('edge', 'invoked')]) {
-    assert.match(r.sql, /part_type = 'tool_call'/, `${r.kind}/${r.type} filters tool_call parts`)
-    assert.match(r.sql, /tool_name IN \('Bash', 'exec_command'\)/, `${r.kind}/${r.type} filters the shell tools`)
+    assert.equal(r.where?.eq?.part_type, 'tool_call', `${r.kind}/${r.type} filters tool_call parts`)
+    assert.deepEqual(r.where?.in?.tool_name, ['Bash', 'exec_command'], `${r.kind}/${r.type} filters the shell tools`)
   }
 })
 
@@ -430,24 +419,26 @@ const MARKER_TEXT = 'Base directory for this skill: /home/u/.claude/skills/hypaw
 const SLASH_TEXT = '<command-name>/hypaware-query</command-name>'
 const CODEX_READ_ARGS = { cmd: 'cat /home/u/.codex/skills/hypaware-query/SKILL.md' }
 
-// @ref LLP 0074#strict-filters [tests]: the SQL filters are the decision, not
+// @ref LLP 0074#strict-filters [tests]: the filters are the decision, not
 // tuning parameters; only role='user' text with a leading anchor is clean.
-test('Skill/ran rules declare the strict per-surface SQL filters', () => {
+// Surfaces 1/4 are declarative predicates (shared scan); surfaces 2/3 stay
+// raw SQL so the content_text prefix guard prunes server-side (LLP 0096).
+test('Skill/ran rules declare the strict per-surface filters', () => {
   for (const kind of /** @type {const} */ (['node', 'edge'])) {
     const type = kind === 'node' ? 'Skill' : 'ran'
     const tool = rule(kind, type, SKILL_TOOL)
-    assert.match(tool.sql, /part_type = 'tool_call' AND tool_name = 'Skill'/, `${kind} surface 1 filters Skill tool calls`)
+    assert.deepEqual(tool.where?.eq, { part_type: 'tool_call', tool_name: 'Skill' }, `${kind} surface 1 filters Skill tool calls`)
 
     const marker = rule(kind, type, SKILL_MARKER)
-    assert.match(marker.sql, /role = 'user' AND part_type = 'text'/, `${kind} surface 2 filters user text parts (assistant-role markers never reach toRow)`)
-    assert.match(marker.sql, /content_text LIKE 'Base directory for this skill: %'/, `${kind} surface 2 anchors the marker in SQL`)
+    assert.match(/** @type {string} */ (marker.sql), /role = 'user' AND part_type = 'text'/, `${kind} surface 2 filters user text parts (assistant-role markers never reach toRow)`)
+    assert.match(/** @type {string} */ (marker.sql), /content_text LIKE 'Base directory for this skill: %'/, `${kind} surface 2 anchors the marker in SQL`)
 
     const slash = rule(kind, type, SKILL_SLASH)
-    assert.match(slash.sql, /role = 'user' AND part_type = 'text'/, `${kind} surface 3 filters user text parts`)
-    assert.match(slash.sql, /content_text LIKE '<command-name>%'/, `${kind} surface 3 anchors the tag in SQL`)
+    assert.match(/** @type {string} */ (slash.sql), /role = 'user' AND part_type = 'text'/, `${kind} surface 3 filters user text parts`)
+    assert.match(/** @type {string} */ (slash.sql), /content_text LIKE '<command-name>%'/, `${kind} surface 3 anchors the tag in SQL`)
 
     const codex = rule(kind, type, SKILL_CODEX)
-    assert.match(codex.sql, /part_type = 'tool_call' AND tool_name = 'exec_command'/, `${kind} surface 4 filters exec_command tool calls`)
+    assert.deepEqual(codex.where?.eq, { part_type: 'tool_call', tool_name: 'exec_command' }, `${kind} surface 4 filters exec_command tool calls`)
   }
 })
 
@@ -592,14 +583,14 @@ test('false-positive matrix: near-miss signals mint nothing', () => {
   assert.equal(codexNode.toRow({ session_id: 's', tool_args: { cmd: 'cat ~/.codex/skills/hypaware-query/reference.md' }, message_created_at: TS }), null, 'reads a different file in the skill dir')
 })
 
-test('aux-tagged rows are excluded from the Skill and ran rules', () => {
+// @ref LLP 0096#decision [tests]: aux exclusion is the contract rowFilter's
+// job for every rule uniformly (the engine applies it before any toRow on
+// both scan paths), so the Skill/ran surfaces need no per-rule aux checks;
+// the rowFilter test above plus the projection e2e cover them.
+test('aux-tagged skill rows are excluded by the contract rowFilter', () => {
+  const { keep } = /** @type {NonNullable<typeof contract.rowFilter>} */ (contract.rowFilter)
   const aux = { attributes: { claude: { aux_kind: 'security_monitor' } } }
-  assert.equal(rule('node', 'Skill', SKILL_TOOL).toRow({ ...aux, session_id: 's', tool_args: { skill: 'x1' }, message_created_at: TS }), null)
-  assert.equal(rule('node', 'Skill', SKILL_MARKER).toRow({ ...aux, session_id: 's', content_text: MARKER_TEXT, message_created_at: TS }), null)
-  assert.equal(rule('node', 'Skill', SKILL_SLASH).toRow({ ...aux, session_id: 's', content_text: SLASH_TEXT, message_created_at: TS }), null)
-  assert.equal(rule('node', 'Skill', SKILL_CODEX).toRow({ ...aux, session_id: 's', tool_args: CODEX_READ_ARGS, message_created_at: TS }), null)
-  assert.equal(rule('edge', 'ran', SKILL_TOOL).toRow({ ...aux, session_id: 's', tool_args: { skill: 'x1' }, message_created_at: TS }), null)
-  assert.equal(rule('edge', 'ran', SKILL_MARKER).toRow({ ...aux, session_id: 's', content_text: MARKER_TEXT, message_created_at: TS }), null)
-  assert.equal(rule('edge', 'ran', SKILL_SLASH).toRow({ ...aux, session_id: 's', content_text: SLASH_TEXT, message_created_at: TS }), null)
-  assert.equal(rule('edge', 'ran', SKILL_CODEX).toRow({ ...aux, session_id: 's', tool_args: CODEX_READ_ARGS, message_created_at: TS }), null)
+  assert.equal(keep({ ...aux, session_id: 's', tool_args: { skill: 'x1' }, message_created_at: TS }), false)
+  assert.equal(keep({ ...aux, session_id: 's', content_text: MARKER_TEXT, message_created_at: TS }), false)
+  assert.equal(keep({ ...aux, session_id: 's', tool_args: CODEX_READ_ARGS, message_created_at: TS }), false)
 })
diff --git a/test/plugins/context-graph-contract.test.js b/test/plugins/context-graph-contract.test.js
index 0bbef80..55e1d77 100644
--- a/test/plugins/context-graph-contract.test.js
+++ b/test/plugins/context-graph-contract.test.js
@@ -126,11 +126,71 @@ test('contract registry rejects malformed rules, naming the offending rule index
   // than mid-projection (or by silently routing rows into the wrong target map).
   assert.throws(() => reg.register(sampleContract({ rules: [ok, { ...ok, kind: 'vertex' }] })), /rule 1 kind/)
   assert.throws(() => reg.register(sampleContract({ rules: [{ ...ok, type: '' }] })), /rule 0 type/)
-  assert.throws(() => reg.register(sampleContract({ rules: [{ ...ok, sql: '' }] })), /rule 0 sql/)
+  assert.throws(() => reg.register(sampleContract({ rules: [{ ...ok, sql: '' }] })), /rule 0 must carry exactly one of sql or columns/)
   assert.throws(() => reg.register(sampleContract({ rules: [{ ...ok, toRow: 'nope' }] })), /rule 0 toRow/)
   assert.throws(() => reg.register(sampleContract({ rules: [null] })), /rule 0 must be an object/)
 })
 
+// @ref LLP 0096#decision [tests]: exactly one read form; where only rides columns; predicate and rowFilter shapes validated at registration
+test('contract registry validates the declarative rule form', () => {
+  const reg = createContractRegistry()
+  const decl = { kind: 'node', type: 'T', columns: ['a'], where: { eq: { b: 'x' } }, toRow: () => null }
+  reg.register(sampleContract({ rules: [decl] }))
+
+  const sql = { kind: 'node', type: 'T', sql: 'SELECT 1', toRow: () => null }
+  assert.throws(() => reg.register(sampleContract({ name: 'y', rules: [{ ...sql, columns: ['a'] }] })), /exactly one of sql or columns/)
+  assert.throws(() => reg.register(sampleContract({ name: 'y', rules: [{ ...sql, where: { eq: { b: 'x' } } }] })), /where is only valid with columns/)
+  assert.throws(() => reg.register(sampleContract({ name: 'y', rules: [{ ...decl, columns: [] }] })), /columns must be non-empty strings/)
+  assert.throws(() => reg.register(sampleContract({ name: 'y', rules: [{ ...decl, where: { gt: { b: 'x' } } }] })), /where.gt is not a supported predicate/)
+  assert.throws(() => reg.register(sampleContract({ name: 'y', rules: [{ ...decl, where: { eq: { b: 7 } } }] })), /where.eq.b must be a non-empty string/)
+  assert.throws(() => reg.register(sampleContract({ name: 'y', rules: [{ ...decl, where: { in: { b: [] } } }] })), /where.in.b must be a non-empty array/)
+})
+
+test('contract registry validates rowFilter, and raw sql must select its columns', () => {
+  const reg = createContractRegistry()
+  const filter = { columns: ['attributes'], keep: () => true }
+  const decl = { kind: 'node', type: 'T', columns: ['a'], toRow: () => null }
+  reg.register(sampleContract({ rules: [decl], rowFilter: filter }))
+
+  assert.throws(() => reg.register(sampleContract({ name: 'y', rules: [decl], rowFilter: { columns: [], keep: () => true } })), /rowFilter columns/)
+  assert.throws(() => reg.register(sampleContract({ name: 'y', rules: [decl], rowFilter: { columns: ['a'] } })), /rowFilter keep/)
+
+  // A raw-SQL rule under a rowFilter must select the filter's columns itself:
+  // the engine cannot inject columns into raw SQL, and keep() over an absent
+  // column would silently pass filtered rows through.
+  const raw = { kind: 'node', type: 'T', sql: 'SELECT a FROM x', toRow: () => null }
+  assert.throws(() => reg.register(sampleContract({ name: 'y', rules: [raw], rowFilter: filter })), /raw sql must select rowFilter column 'attributes'/)
+  reg.register(sampleContract({ name: 'z', rules: [{ ...raw, sql: 'SELECT attributes, a FROM x' }], rowFilter: filter }))
+
+  // The guard checks the SELECT projection list, not a loose substring: the
+  // column named in a WHERE clause, or a different column whose name merely
+  // contains it, is not projected and must be rejected.
+  const notProjected = [
+    'SELECT a FROM x WHERE attributes IS NOT NULL',
+    'SELECT other_attributes FROM x',
+    'SELECT a FROM x WHERE b IN (SELECT attributes FROM y)',
+  ]
+  for (const sql of notProjected) {
+    assert.throws(
+      () => reg.register(sampleContract({ name: 'n', rules: [{ ...raw, sql }], rowFilter: filter })),
+      /raw sql must select rowFilter column 'attributes'/,
+      sql
+    )
+  }
+
+  // Projection forms that do provably carry the column are accepted: a
+  // qualified reference, an alias, and a wildcard.
+  const projected = [
+    'SELECT x.attributes, a FROM x',
+    'SELECT foo AS attributes, a FROM x',
+    'SELECT * FROM x',
+    'SELECT x.* FROM x',
+  ]
+  projected.forEach((sql, i) => {
+    reg.register(sampleContract({ name: `p${i}`, rules: [{ ...raw, sql }], rowFilter: filter }))
+  })
+})
+
 test('contract registry rejects a duplicate (plugin, name)', () => {
   const reg = createContractRegistry()
   reg.register(sampleContract())
diff --git a/test/plugins/context-graph-project-e2e.test.js b/test/plugins/context-graph-project-e2e.test.js
index 2654665..e54db03 100644
--- a/test/plugins/context-graph-project-e2e.test.js
+++ b/test/plugins/context-graph-project-e2e.test.js
@@ -425,3 +425,53 @@ test('graph neighbors traversal: --edge-type ran reaches Skill, --edge-type invo
     )
   }, SKILL_PROGRAM_ROWS)
 })
+
+/**
+ * Rebuild a contract's declarative rules as the raw SQL the engine used to
+ * run per rule, `attributes` prepended exactly like the old per-rule aux
+ * wrap. Raw-SQL rules pass through untouched. Running both variants over one
+ * fixture pins the JS predicate evaluator to the SQL engine's semantics.
+ *
+ * @param {Contract} contract
+ * @returns {Contract}
+ */
+function sqlTwin(contract) {
+  const rules = contract.rules.map((rule) => {
+    if (typeof rule.sql === 'string') return rule
+    const columns = ['attributes', ...(rule.columns ?? [])]
+    /** @type {string[]} */
+    const clauses = []
+    for (const [col, value] of Object.entries(rule.where?.eq ?? {})) clauses.push(`${col} = '${value}'`)
+    for (const [col, values] of Object.entries(rule.where?.in ?? {})) clauses.push(`${col} IN (${values.map((v) => `'${v}'`).join(', ')})`)
+    for (const [col, prefix] of Object.entries(rule.where?.likePrefix ?? {})) clauses.push(`${col} LIKE '${prefix}%'`)
+    const where = clauses.length > 0 ? ` WHERE ${clauses.join(' AND ')}` : ''
+    return { ...rule, columns: undefined, where: undefined, sql: `SELECT ${columns.join(', ')} FROM ${contract.sourceDataset}${where}` }
+  })
+  return { ...contract, rules }
+}
+
+// @ref LLP 0096#decision [tests]: the shared scan + JS predicates produce
+// bit-identical graph rows to the per-rule SQL execution they replaced,
+// over a fixture that exercises every predicate shape (eq, in, likePrefix)
+// plus the aux rowFilter and null columns.
+test('shared-scan projection is row-identical to per-rule SQL execution', async () => {
+  const fixture = [...ROWS, AUX_ROW, ...SKILL_PROGRAM_ROWS]
+  /** @type {{ ids: string[], rows: Record<string, unknown>[] }[]} */
+  const outcomes = []
+  for (const variant of ['declarative', 'sql-twin']) {
+    await withSeededGateway(async ({ registry, storage }) => {
+      const contract = createAiGatewayGraphContract({ nodeId, edgeId, makeRowBuilders })
+      const run = variant === 'declarative' ? contract : sqlTwin(contract)
+      const report = await projectGraph({ query: registry, storage, contracts: [run] })
+      assert.ok(report.nodesWritten > 0, `${variant}: fixture mints nodes`)
+      const nodes = await executeQuerySql({ query: 'SELECT node_id, node_type, natural_key, label, props FROM node', registry, storage, refresh: 'always' })
+      const edges = await executeQuerySql({ query: 'SELECT edge_id, src_id, dst_id, edge_type, props FROM edge', registry, storage, refresh: 'always' })
+      const key = (/** @type {Record<string, unknown>} */ r) => JSON.stringify(r)
+      outcomes.push({
+        ids: [...nodes.rows, ...edges.rows].map(key).sort(),
+        rows: [...nodes.rows, ...edges.rows],
+      })
+    }, fixture)
+  }
+  assert.deepEqual(outcomes[0].ids, outcomes[1].ids, 'identical node/edge rows from both execution paths')
+})
diff --git a/test/plugins/context-graph-project.test.js b/test/plugins/context-graph-project.test.js
index fe5081d..5fd3279 100644
--- a/test/plugins/context-graph-project.test.js
+++ b/test/plugins/context-graph-project.test.js
@@ -3,7 +3,7 @@
 import assert from 'node:assert/strict'
 import test from 'node:test'
 
-import { firstSeenTime, mergeRow } from '../../hypaware-core/plugins-workspace/context-graph/src/project.js'
+import { firstSeenTime, matchesPredicate, mergeRow } from '../../hypaware-core/plugins-workspace/context-graph/src/project.js'
 
 /**
  * @param {Partial<Record<string, unknown>>} overrides
@@ -93,6 +93,23 @@ test('mergeRow tolerates rows with unparseable or missing first_seen', () => {
   assert.equal(ba.first_seen, '2026-06-01T00:00:00.000Z')
 })
 
+// The equivalence e2e (context-graph-project-e2e.test.js) claims likePrefix
+// coverage, but the ai-gateway prefix surfaces run as raw SQL, so its JS
+// `likePrefix` branch is never actually evaluated there. Exercise it directly.
+// @ref LLP 0096#decision [tests]: likePrefix matches a string prefix; null and non-string columns never match
+test('matchesPredicate likePrefix keeps only matching-prefix string rows', () => {
+  const where = { likePrefix: { content_text: 'Base directory for this skill: ' } }
+  const match = { content_text: 'Base directory for this skill: /repo/skills/foo' }
+  const nonMatch = { content_text: 'a different message' }
+  const nullValue = { content_text: null }
+  const nonString = { content_text: 42 }
+
+  assert.equal(matchesPredicate(where, match), true, 'matching prefix is kept')
+  assert.equal(matchesPredicate(where, nonMatch), false, 'non-matching string is dropped')
+  assert.equal(matchesPredicate(where, nullValue), false, 'null never matches')
+  assert.equal(matchesPredicate(where, nonString), false, 'non-string never matches')
+})
+
 test('firstSeenTime normalizes strings, Dates, and epoch numbers', () => {
   const t = Date.UTC(2026, 5, 1)
   assert.equal(firstSeenTime('2026-06-01T00:00:00.000Z'), t)