diff --git a/compiler/bin/main.ml b/compiler/bin/main.ml
index a8f1276..a3240bb 100644
--- a/compiler/bin/main.ml
+++ b/compiler/bin/main.ml
@@ -600,10 +600,23 @@ let manifest_parse (path : string) : (string * string) list =
        if line = "" || (String.length line >= 1 && line.[0] = '#') then ()
        else if line.[0] = '[' then begin
          if line.[String.length line - 1] <> ']' then fail !lineno "malformed section header";
-         section := String.sub line 1 (String.length line - 2);
-         if !section <> "runtime" && !section <> "build" then
-           fail !lineno (Printf.sprintf "unknown section [%s] (runtime and build exist)" !section)
+         (* accept `[[table.array]]` headers too (iteration 9c's
+            [[share.clients]]) by trimming the doubled brackets *)
+         let inner = String.sub line 1 (String.length line - 2) in
+         let inner =
+           if String.length inner >= 2 && inner.[0] = '[' && inner.[String.length inner - 1] = ']'
+           then String.sub inner 1 (String.length inner - 2)
+           else inner
+         in
+         section := inner;
+         if !section <> "runtime" && !section <> "build" && !section <> "share"
+            && !section <> "share.clients"
+         then
+           fail !lineno
+             (Printf.sprintf "unknown section [%s] (runtime and build exist)" !section)
        end
+       else if !section = "share" || !section = "share.clients" then
+         () (* iteration 9c manifest keys — parsed by the attach feature, ignored here *)
        else
          match String.index_opt line '=' with
          | None -> fail !lineno "expected `key = \"value\"`"
diff --git a/compiler/src/ast.ml b/compiler/src/ast.ml
index 91ffca9..840f2f0 100644
--- a/compiler/src/ast.ml
+++ b/compiler/src/ast.ml
@@ -69,6 +69,9 @@ type field_ty =
   | Ref of string
   | Multi of string
   | Map of string * string (* key type, value type: map<K, V> *)
+  | Backlink of string * string (* backlink C.f: the computed inverse of a
+                                    `ref` — NOT a stored column; reading it
+                                    scans C's index on f. Types as multi C. *)
   | Nullable of field_ty   (* ?T wrapper *)
 
 (* Parameter passing convention (spec section 3, rule 2): default is an
@@ -216,6 +219,10 @@ and expr_kind =
      literal, returns the new row's id (Int), legal in statement and
      expression position both. `select` stays a DbStub until Task 5. *)
   | Insert of string * (string * expr) list
+  (* `delete <row>` (iteration 9b): removes the row a table-class value
+     names; an expression yielding the deleted id (restrict/trap surfaces
+     through the engine like any DB fault, catchable). *)
+  | Delete of expr
   (* haxe-parity Task 2: one `${expr}` interpolation site, produced only
      by the string-interpolation desugar (parser.ml) — never written
      directly by a parse rule the way every other expr_kind is. Its
@@ -279,6 +286,31 @@ and expr_kind =
       ename : string;
       handler : stmt list;
     }
+  (* iteration 9b: a language-integrated query. `from <var> in <source>
+     where <e>* [group <e> by <k> into <g>] [order by <e> [desc]] [take <e>]
+     select <e>` — lowered to a bytecode loop over engine cursor builtins,
+     never SQL text. A table-class value is its row id at runtime, so field
+     access on a range variable reads through the engine. Slice scope today:
+     from/where/order/take/select and group-by aggregation; join is later. *)
+  | Query of query
+
+and query_source =
+  | QTable of string (* a table class by name: `from e in Employee` *)
+  | QNav of expr (* a backlink/multi navigation: `from s in d.staff` *)
+
+and query = {
+  q_var : string;
+  q_src : query_source;
+  q_wheres : expr list;
+  (* group <key_expr> by ... into <gvar>: present iff this is an aggregating
+     query. q_group_key is the whole grouped element (`e`), q_group_by the
+     key, q_gvar the group binding whose `.f` columns feed aggregates. *)
+  q_group : (string * expr) option; (* (gvar, key_expr) *)
+  q_order : (expr * bool) option;    (* (key, desc?) *)
+  q_take : expr option;
+  q_select : expr;
+  q_pos : pos;
+}
 
 (* ---- statements (Task 5) ---------------------------------------------
 
diff --git a/compiler/src/dump.ml b/compiler/src/dump.ml
index 9914646..3cfe237 100644
--- a/compiler/src/dump.ml
+++ b/compiler/src/dump.ml
@@ -150,6 +150,7 @@ let rec field_ty_str : Ast.field_ty -> string = function
   | Ast.Ref s -> Printf.sprintf "ref %s" s
   | Ast.Multi s -> Printf.sprintf "multi %s" s
   | Ast.Map (k, v) -> Printf.sprintf "map<%s, %s>" k v
+  | Ast.Backlink (c, f) -> Printf.sprintf "backlink %s.%s" c f
   | Ast.Nullable t -> "?" ^ field_ty_str t
 
 let param_str (p : Ast.param) : string = Printf.sprintf "%s%s: %s" (conv_str p.conv) p.name (field_ty_str p.ty)
@@ -226,6 +227,15 @@ let rec expr_str (e : Ast.expr) : string =
     Printf.sprintf "INSERT %s { %s }" name
       (String.concat ", "
          (List.map (fun (fname, fval) -> Printf.sprintf "%s: %s" fname (expr_str fval)) fields))
+  | Ast.Query q ->
+    let src = match q.Ast.q_src with Ast.QTable cn -> cn | Ast.QNav e -> expr_str e in
+    Printf.sprintf "QUERY from %s in %s%s%s select %s" q.Ast.q_var src
+      (String.concat "" (List.map (fun w -> " where " ^ expr_str w) q.Ast.q_wheres))
+      (match q.Ast.q_group with
+       | Some (g, k) -> Printf.sprintf " group by %s into %s" (expr_str k) g
+       | None -> "")
+      (expr_str q.Ast.q_select)
+  | Ast.Delete t -> Printf.sprintf "DELETE %s" (expr_str t)
   | Ast.DbStub toks -> Printf.sprintf "DB_STUB(%s)" (dbstub_tokens_str toks)
   | Ast.Interp inner -> Printf.sprintf "INTERP(%s)" (expr_str inner)
   | Ast.ListLit items -> Printf.sprintf "[%s]" (String.concat ", " (List.map expr_str items))
diff --git a/compiler/src/emit.ml b/compiler/src/emit.ml
index 136b40b..5dcf173 100644
--- a/compiler/src/emit.ml
+++ b/compiler/src/emit.ml
@@ -354,6 +354,11 @@ type clsrec = {
      unique single-column entry per `@unique` field. Serialized as the v3
      class-record tail; the engine builds its runtime indexes from this. *)
   cr_indexes : (bool * int array) list;
+  cr_is_table : bool; (* has @table — its instances are row ids (iteration 9b) *)
+  (* backlink fields (iteration 9b): name -> (source class, source field).
+     Virtual — not in cr_fields, no stored column; `d.staff` reads them by
+     probing the source class's index on the source field. *)
+  cr_backlinks : (string * (string * string)) list;
 }
 
 type ifacerec = {
@@ -778,6 +783,44 @@ let field_kind (p : pctx) (ft : Ast.field_ty) : int =
 
 let class_of_name (p : pctx) (n : string) : int option = SM.find_opt n p.p_class_id
 
+(* iteration 9b: a @table class's instances are row ids, so field access on
+   one reads through the engine (DB_GET_FIELD) rather than GETF. *)
+let is_table_class (p : pctx) (cid : int) : bool =
+  cid >= 0 && cid < Array.length p.p_classes && p.p_classes.(cid).cr_is_table
+
+let b_str_lt = 67
+let b_db_update_field = 62
+let b_db_delete = 63
+let b_db_scan = 64
+let b_db_get_field = 65
+let b_db_probe = 66
+
+(* iteration 9b: `d.staff` where staff is `backlink Employee.dept` reads by
+   probing Employee's index on its `dept` column. Resolve to (source cid,
+   index number) — None if the source field is not a declared index (a
+   backlink without a backing index has no efficient read and is rejected). *)
+let backlink_target (p : pctx) (base_cid : int) (fname : string) : (int * int) option =
+  match List.assoc_opt fname p.p_classes.(base_cid).cr_backlinks with
+  | None -> None
+  | Some (src_class, src_field) -> (
+    match class_of_name p src_class with
+    | None -> None
+    | Some scid ->
+      let sc = p.p_classes.(scid) in
+      (* stored column index of the source field *)
+      let col = ref (-1) in
+      Array.iteri (fun i (n, _) -> if n = src_field then col := i) sc.cr_fields;
+      if !col < 0 then None
+      else
+        (* the index whose single column is that field *)
+        let rec find n = function
+          | [] -> None
+          | (_, cols) :: tl ->
+            if Array.length cols = 1 && cols.(0) = !col then Some (scid, n)
+            else find (n + 1) tl
+        in
+        find 0 sc.cr_indexes)
+
 let field_of (p : pctx) (cid : int) (fname : string) : (int * Ast.field_ty) option =
   let fs = p.p_classes.(cid).cr_fields in
   let rec go i = if i >= Array.length fs then None else
@@ -947,6 +990,22 @@ let variant_tag_value (p : pctx) (u : Types.union_info) (vi : Types.variant_info
     | None -> 0 (* unreachable: pass 1 registers every payload-union variant *)
   else vi.Types.vi_tag
 
+(* iteration 9b: a query's element type, as the name a `Multi` carries.
+   `select x` yields the source class (a row id typed as the class);
+   `select x.field` yields that field's type; anything else falls back to
+   Int (the slice's shapes are these two). *)
+let query_elem_scalar (p : pctx) (q : Ast.query) ~(src : string) : string =
+  match q.Ast.q_select.Ast.kind with
+  | Ast.Ident v when v = q.Ast.q_var -> src (* select the whole row: element = source class *)
+  | Ast.Field ({ Ast.kind = Ast.Ident v; _ }, fname) when v = q.Ast.q_var -> (
+    match class_of_name p src with
+    | Some cid -> (
+      match field_of p cid fname with
+      | Some (_, ty) -> ( match unwrap ty with Scalar n -> n | _ -> "Int")
+      | None -> "Int")
+    | None -> "Int")
+  | _ -> "Int"
+
 let rec ty_of_expr (p : pctx) (f : fstate) (e : Ast.expr) : Ast.field_ty option =
   match e.kind with
   | IntLit _ -> Some (Scalar "Int")
@@ -976,10 +1035,17 @@ let rec ty_of_expr (p : pctx) (f : fstate) (e : Ast.expr) : Ast.field_ty option
   | Field (base, fname) -> (
     match ty_of_expr p f base with
     | Some bt -> (
-      match unwrap bt with
+      (* a `ref C` navigates into C: the target is a table row id *)
+      match (match unwrap bt with Ref c -> Scalar c | other -> other) with
       | Scalar cn -> (
         match class_of_name p cn with
-        | Some cid -> ( match field_of p cid fname with Some (_, t) -> Some t | None -> None)
+        | Some cid -> (
+          match field_of p cid fname with
+          | Some (_, t) -> Some t
+          | None -> (
+            match List.assoc_opt fname p.p_classes.(cid).cr_backlinks with
+            | Some (sc, _) -> Some (Multi sc)
+            | None -> None))
         | None -> None)
       | _ -> None)
     | None -> None)
@@ -1061,6 +1127,15 @@ let rec ty_of_expr (p : pctx) (f : fstate) (e : Ast.expr) : Ast.field_ty option
     | Add | Sub | Mul | Div | Mod -> ( match ty_of_expr p f l with Some t -> Some t | None -> Some (Scalar "Int")))
   | Ctor (cn, _) -> Some (Scalar cn)
   | Insert _ -> Some (Scalar "Int")
+  | Delete _ -> Some (Scalar "Int")
+  | Query q ->
+    let src =
+      match q.Ast.q_src with
+      | Ast.QTable cn -> cn
+      | Ast.QNav nav -> (
+        match ty_of_expr p f nav with Some t -> (match unwrap t with Multi c -> c | Scalar c -> c | _ -> "") | None -> "")
+    in
+    Some (Multi (query_elem_scalar p q ~src))
   | Interp _ -> Some (Scalar "Text")
   | DbStub _ -> None
   | Switch (subject, arms) -> (
@@ -1351,7 +1426,8 @@ let field_class_meta (p : pctx) (ty : Ast.field_ty) : int =
     match name_of (Ast.Scalar e) with
     | Some n -> ( match class_of_name p n with Some cid -> cid | None -> wob_none)
     | None -> wob_none)
-  | Ast.Ref _ | Ast.Nullable _ -> wob_none
+  | Ast.Ref n -> ( match class_of_name p n with Some cid -> cid | None -> wob_none)
+  | Ast.Backlink _ | Ast.Nullable _ -> wob_none
 
 let field_elem_meta (p : pctx) (ty : Ast.field_ty) : int =
   match unwrap ty with
@@ -1590,9 +1666,22 @@ let rec emit_expr (p : pctx) (f : fstate) (v : views) ~(dst : int) ?expected (e
   | Field (base, fname) -> (
     match ty_of_expr p f base with
     | Some bt -> (
-      match unwrap bt with
+      match (match unwrap bt with Ref c -> Scalar c | other -> other) with
       | Scalar cn -> (
         match class_of_name p cn with
+        | Some cid when is_table_class p cid && backlink_target p cid fname <> None -> (
+          (* `d.staff`: probe the source class's index for rows referencing
+             this row's id. Window: [class, index, key(=base id)]. *)
+          match backlink_target p cid fname with
+          | Some (scid, ino) ->
+            let b = emit_operand p f v base in
+            let w = alloc_temps p f e.pos 3 in
+            put f (ins_abx op_loadk w (check_bx p f e.pos "constant" (const_int p scid)));
+            put f (ins_abx op_loadk (w + 1) (check_bx p f e.pos "constant" (const_int p ino)));
+            put f (ins_abc op_move (w + 2) b 0);
+            sync_mask p f v e.id;
+            put f (ins_abc op_builtin dst w b_db_probe)
+          | None -> ())
         | Some cid -> (
           match field_of p cid fname with
           | Some (idx, _) ->
@@ -1610,7 +1699,17 @@ let rec emit_expr (p : pctx) (f : fstate) (v : views) ~(dst : int) ?expected (e
               f.f_stmt_drops <- g :: f.f_stmt_drops;
               f.f_esc_drops <- g :: f.f_esc_drops
             end;
-            put f (ins_abc op_getf dst b (check_field_idx p f e.pos idx))
+            if is_table_class p cid then begin
+              (* a table-class value is its row id; read the column from the
+                 engine. Window: [class-id, id, field-idx]. *)
+              let w = alloc_temps p f e.pos 3 in
+              put f (ins_abx op_loadk w (check_bx p f e.pos "constant" (const_int p cid)));
+              put f (ins_abc op_move (w + 1) b 0);
+              put f (ins_abx op_loadk (w + 2) (check_bx p f e.pos "constant" (const_int p idx)));
+              sync_mask p f v e.id;
+              put f (ins_abc op_builtin dst w b_db_get_field)
+            end
+            else put f (ins_abc op_getf dst b (check_field_idx p f e.pos idx))
           | None ->
             err p ~code:cannot_lower_code ~file:f.f_file ~pos:e.pos
               ~message:(Printf.sprintf "`%s` has no field `%s`" cn fname);
@@ -1662,6 +1761,42 @@ let rec emit_expr (p : pctx) (f : fstate) (v : views) ~(dst : int) ?expected (e
   | Binary (op, l, r) -> emit_binary p f v ~dst op l r
   | Ctor (cn, fields) -> emit_ctor p f v ~dst e cn fields
   | Insert (cn, fields) -> emit_insert p f v ~dst e cn fields
+  | Delete target -> (
+    match ty_of_expr p f target with
+    | Some bt -> (
+      match (match unwrap bt with Ref c -> Scalar c | o -> o) with
+      | Scalar cn -> (
+        match class_of_name p cn with
+        | Some cid when is_table_class p cid ->
+          (* reserve dst past the window: in tail position dst == the first
+             window reg, and moving the id into dst would clobber the class
+             id — the disassembly-caught bug *)
+          let outer = f.f_temp in
+          if f.f_temp <= dst then f.f_temp <- dst + 1;
+          let w = alloc_temps p f e.pos 2 in
+          put f (ins_abx op_loadk w (check_bx p f e.pos "constant" (const_int p cid)));
+          let save = f.f_temp in
+          emit_expr p f v ~dst:(w + 1) target;
+          f.f_temp <- save;
+          (* keep the id so `delete x` can be used as an expression *)
+          put f (ins_abc op_move dst (w + 1) 0);
+          sync_mask p f v e.id;
+          f.f_cur_line <- e.pos.line;
+          put f (ins_abc op_builtin w w b_db_delete);
+          f.f_temp <- outer
+        | _ ->
+          err p ~code:cannot_lower_code ~file:f.f_file ~pos:e.pos
+            ~message:"`delete` target is not a table row";
+          put f (ins_abx op_loadk dst (const_int p 0)))
+      | _ ->
+        err p ~code:cannot_lower_code ~file:f.f_file ~pos:e.pos
+          ~message:"`delete` target is not a table row";
+        put f (ins_abx op_loadk dst (const_int p 0)))
+    | None ->
+      err p ~code:cannot_lower_code ~file:f.f_file ~pos:e.pos
+        ~message:"cannot resolve the `delete` target's type";
+      put f (ins_abx op_loadk dst (const_int p 0)))
+  | Query q -> emit_query p f v ~dst e q
   | Interp inner -> (
     (* haxe-parity Task 2: the type-directed half of the interpolation
        desugar (parser.ml's own doc comment on Ast.Interp) — a Text
@@ -2364,6 +2499,256 @@ and emit_ctor (p : pctx) (f : fstate) (v : views) ~(dst : int) (e : Ast.expr) (c
    nullable scalar, the zero word otherwise). The engine COPIES every
    value at the row API, so after the builtin every freshly built
    argument is still this frame's to drop — same reap as push/set. *)
+and emit_query (p : pctx) (f : fstate) (v : views) ~(dst : int) (e : Ast.expr)
+    (q : Ast.query) : unit =
+  (* iteration 9b slice: from/where/select over a table scan. group/order/
+     take/navigation are diagnosed in types.ml, so a written image never
+     reaches this with them set. Lowered to an ordinary bytecode loop over
+     DB_SCAN's materialized id list — no plan tree, no text. *)
+  let cn =
+    match q.Ast.q_src with
+    | Ast.QTable cn -> cn
+    | Ast.QNav nav -> (
+      (* the source's element type is the range var's class *)
+      match ty_of_expr p f nav with
+      | Some t -> ( match unwrap t with Multi c -> c | Scalar c -> c | _ -> "")
+      | None -> "")
+  in
+  match class_of_name p cn with
+  | None ->
+    err p ~code:cannot_lower_code ~file:f.f_file ~pos:e.pos
+      ~message:(Printf.sprintf "query over `%s`, which is not a declared table class" cn);
+    put f (ins_abx op_loadk dst (const_int p 0))
+  | Some cid ->
+    let elem_name = query_elem_scalar p q ~src:cn in
+    let elem = Scalar elem_name in
+    (* a table-class element is a row ID (a scalar), not a heap pointer — so
+       the result container is SCALAR-kinded even though the element TYPES as
+       the class; getting this wrong drops an id as a pointer (ASan SEGV) *)
+    let elem_kind =
+      match class_of_name p elem_name with
+      | Some ecid when is_table_class p ecid -> 0 (* WO_K_SCALAR *)
+      | _ -> field_kind p elem
+    in
+    (* reserve dst past the loop's working registers (same guard emit_ctor
+       uses): dst holds the result multi every push writes into *)
+    let outer = f.f_temp in
+    if f.f_temp <= dst then f.f_temp <- dst + 1;
+    (* loop-carried registers, allocated once above dst, never reset *)
+    let scan = alloc_temp p f e.pos in
+    let idx = alloc_temp p f e.pos in
+    let len = alloc_temp p f e.pos in
+    let idreg = alloc_temp p f e.pos in
+    let body_base = f.f_temp in
+    (* scan -> multi of ids; result multi -> dst *)
+    sync_mask p f v e.id;
+    f.f_cur_line <- e.pos.line;
+    (match q.Ast.q_src with
+    | Ast.QTable _ ->
+      put f (ins_abx op_loadk scan (check_bx p f e.pos "constant" (const_int p cid)));
+      put f (ins_abc op_builtin scan scan b_db_scan)
+    | Ast.QNav nav ->
+      (* the navigation (a backlink) already yields a multi of source ids *)
+      let save = f.f_temp in
+      f.f_temp <- scan + 1;
+      emit_expr p f v ~dst:scan nav;
+      f.f_temp <- save);
+    put f (ins_abc op_builtin dst elem_kind b_multi_new);
+    put f (ins_abc op_builtin len scan b_len);
+    put f (ins_abx op_loadk idx (check_bx p f e.pos "constant" (const_int p 0)));
+    (* bind the range var to the current id (typed as the class), so field
+       access inside where/select routes through DB_GET_FIELD *)
+    let saved_env = f.f_env in
+    f.f_env <- (q.Ast.q_var, (idreg, Scalar cn)) :: f.f_env;
+    ignore body_base;
+    let top = here f in
+    f.f_temp <- body_base;
+    let tc = alloc_temp p f e.pos in
+    put f (ins_abc op_lt tc idx len);
+    let jz_exit = here f in
+    put f (ins_asbx op_jz tc 0);
+    (* id = multi_get(scan, idx) *)
+    let w = alloc_temps p f e.pos 2 in
+    put f (ins_abc op_move w scan 0);
+    put f (ins_abc op_move (w + 1) idx 0);
+    put f (ins_abc op_builtin idreg w b_multi_get);
+    (* where guards: any false skips the push *)
+    let skips = ref [] in
+    List.iter
+      (fun w_expr ->
+        let save = f.f_temp in
+        let wr = emit_operand p f v w_expr in
+        skips := here f :: !skips;
+        put f (ins_asbx op_jz wr 0);
+        f.f_temp <- save)
+      q.Ast.q_wheres;
+    (* select -> push into dst (copying a Text element the container owns) *)
+    let save = f.f_temp in
+    let sel = alloc_temp p f e.pos in
+    emit_expr p f v ~dst:sel q.Ast.q_select;
+    if elem_kind = 3 then put f (ins_abc op_builtin sel sel b_text_copy);
+    let pw = alloc_temps p f e.pos 2 in
+    put f (ins_abc op_move pw dst 0);
+    put f (ins_abc op_move (pw + 1) sel 0);
+    put f (ins_abc op_builtin pw pw b_multi_push);
+    f.f_temp <- save;
+    (* skip target: increment and loop *)
+    let cont = here f in
+    List.iter (fun pc -> patch_jump p f ~file:f.f_file ~pos:e.pos pc cont) !skips;
+    f.f_temp <- body_base;
+    let one = alloc_temp p f e.pos in
+    put f (ins_abx op_loadk one (check_bx p f e.pos "constant" (const_int p 1)));
+    put f (ins_abc op_add idx idx one);
+    let back = here f in
+    put f (ins_asbx op_jmp 0 0);
+    patch_jump p f ~file:f.f_file ~pos:e.pos back top;
+    let exit_pc = here f in
+    patch_jump p f ~file:f.f_file ~pos:e.pos jz_exit exit_pc;
+    f.f_env <- saved_env;
+    (* the scan's id list was this query's own, dropped now *)
+    put f (ins_abc op_drop scan 0 0);
+    (* ---- order by (whole-row selection sort) ----------------------------
+       Elements of dst are row ids; the key re-reads a field through the
+       range var. Selection sort is O(n^2) but the result sets here are
+       small and this is KISS by design (no cost planner). Only the
+       whole-row + field-key shape is supported; grouped/projection ordering
+       lands with group-by. *)
+    (match q.Ast.q_order with
+    | Some (key, desc) ->
+      f.f_temp <- body_base;
+      let n = alloc_temp p f e.pos in
+      put f (ins_abc op_builtin n dst b_count);
+      let i = alloc_temp p f e.pos in
+      let j = alloc_temp p f e.pos in
+      let best = alloc_temp p f e.pos in
+      let elem_j = alloc_temp p f e.pos in
+      let elem_b = alloc_temp p f e.pos in
+      let sort_scratch = f.f_temp in
+      put f (ins_abx op_loadk i (check_bx p f e.pos "constant" (const_int p 0)));
+      let oi = here f in (* outer: while i < n *)
+      let oc = alloc_temp p f e.pos in
+      put f (ins_abc op_lt oc i n);
+      let ojz = here f in
+      put f (ins_asbx op_jz oc 0);
+      put f (ins_abc op_move best i 0);
+      let oneA = alloc_temp p f e.pos in
+      put f (ins_abx op_loadk oneA (check_bx p f e.pos "constant" (const_int p 1)));
+      put f (ins_abc op_add j i oneA);
+      let ij = here f in (* inner: while j < n *)
+      let ic = alloc_temp p f e.pos in
+      put f (ins_abc op_lt ic j n);
+      let ijz = here f in
+      put f (ins_asbx op_jz ic 0);
+      (* elem_j = multi_get(dst,j); elem_b = multi_get(dst,best) *)
+      let gw = alloc_temps p f e.pos 2 in
+      put f (ins_abc op_move gw dst 0);
+      put f (ins_abc op_move (gw + 1) j 0);
+      put f (ins_abc op_builtin elem_j gw b_multi_get);
+      put f (ins_abc op_move (gw + 1) best 0);
+      put f (ins_abc op_builtin elem_b gw b_multi_get);
+      (* keys: bind range var to elem_j / elem_b, eval key expr *)
+      let saved_env2 = f.f_env in
+      f.f_temp <- sort_scratch;
+      f.f_env <- (q.Ast.q_var, (elem_j, Scalar cn)) :: saved_env2;
+      (* key kind must be read with the range var BOUND — else ty_of_expr of
+         `x.name` sees x unbound, returns None, and a Text key silently falls
+         to the pointer-comparing op_lt (the wrong-order bug) *)
+      let key_is_text =
+        match ty_of_expr p f key with Some t -> field_kind p t = 3 | None -> false
+      in
+      let kj = alloc_temp p f e.pos in
+      emit_expr p f v ~dst:kj key;
+      f.f_env <- (q.Ast.q_var, (elem_b, Scalar cn)) :: saved_env2;
+      let kb = alloc_temp p f e.pos in
+      emit_expr p f v ~dst:kb key;
+      f.f_env <- saved_env2;
+      (* cmp: for asc, kj < kb -> best=j; for desc, kj > kb (== kb < kj). *)
+      let cmp = alloc_temp p f e.pos in
+      let lt a b =
+        if key_is_text then begin
+          let save = f.f_temp in
+          let w = alloc_temps p f e.pos 2 in
+          put f (ins_abc op_move w a 0);
+          put f (ins_abc op_move (w + 1) b 0);
+          put f (ins_abc op_builtin cmp w b_str_lt);
+          f.f_temp <- save
+        end
+        else put f (ins_abc op_lt cmp a b)
+      in
+      if desc then lt kb kj else lt kj kb;
+      let cjz = here f in
+      put f (ins_asbx op_jz cmp 0);
+      put f (ins_abc op_move best j 0);
+      let after = here f in
+      patch_jump p f ~file:f.f_file ~pos:e.pos cjz after;
+      f.f_temp <- sort_scratch;
+      let oneB = alloc_temp p f e.pos in
+      put f (ins_abx op_loadk oneB (check_bx p f e.pos "constant" (const_int p 1)));
+      put f (ins_abc op_add j j oneB);
+      let iback = here f in
+      put f (ins_asbx op_jmp 0 0);
+      patch_jump p f ~file:f.f_file ~pos:e.pos iback ij;
+      let iexit = here f in
+      patch_jump p f ~file:f.f_file ~pos:e.pos ijz iexit;
+      (* swap dst[i], dst[best]: read both, multi_set both *)
+      f.f_temp <- sort_scratch;
+      let vi = alloc_temp p f e.pos in
+      let vb = alloc_temp p f e.pos in
+      let sw = alloc_temps p f e.pos 3 in
+      put f (ins_abc op_move sw dst 0);
+      put f (ins_abc op_move (sw + 1) i 0);
+      put f (ins_abc op_builtin vi sw b_multi_get);
+      put f (ins_abc op_move (sw + 1) best 0);
+      put f (ins_abc op_builtin vb sw b_multi_get);
+      (* dst[i] = vb *)
+      put f (ins_abc op_move sw dst 0);
+      put f (ins_abc op_move (sw + 1) i 0);
+      put f (ins_abc op_move (sw + 2) vb 0);
+      put f (ins_abc op_builtin sw sw b_multi_set);
+      (* dst[best] = vi *)
+      put f (ins_abc op_move sw dst 0);
+      put f (ins_abc op_move (sw + 1) best 0);
+      put f (ins_abc op_move (sw + 2) vi 0);
+      put f (ins_abc op_builtin sw sw b_multi_set);
+      f.f_temp <- sort_scratch;
+      let oneC = alloc_temp p f e.pos in
+      put f (ins_abx op_loadk oneC (check_bx p f e.pos "constant" (const_int p 1)));
+      put f (ins_abc op_add i i oneC);
+      let oback = here f in
+      put f (ins_asbx op_jmp 0 0);
+      patch_jump p f ~file:f.f_file ~pos:e.pos oback oi;
+      let oexit = here f in
+      patch_jump p f ~file:f.f_file ~pos:e.pos ojz oexit
+    | None -> ());
+    (* ---- take N: slice dst to [0, N) --------------------------------- *)
+    (match q.Ast.q_take with
+    | Some tk ->
+      f.f_temp <- body_base;
+      let nreg = alloc_temp p f e.pos in
+      emit_expr p f v ~dst:nreg tk;
+      (* clamp N to count(dst) so slice never runs past the end *)
+      let cnt = alloc_temp p f e.pos in
+      put f (ins_abc op_builtin cnt dst b_count);
+      let over = alloc_temp p f e.pos in
+      put f (ins_abc op_lt over cnt nreg); (* count < N ? use count *)
+      let jz2 = here f in
+      put f (ins_asbx op_jz over 0);
+      put f (ins_abc op_move nreg cnt 0);
+      let aft = here f in
+      patch_jump p f ~file:f.f_file ~pos:e.pos jz2 aft;
+      let sw = alloc_temps p f e.pos 3 in
+      let zero = alloc_temp p f e.pos in
+      put f (ins_abx op_loadk zero (check_bx p f e.pos "constant" (const_int p 0)));
+      put f (ins_abc op_move sw dst 0);
+      put f (ins_abc op_move (sw + 1) zero 0);
+      put f (ins_abc op_move (sw + 2) nreg 0);
+      let sliced = alloc_temp p f e.pos in
+      put f (ins_abc op_builtin sliced sw b_slice);
+      put f (ins_abc op_drop dst 0 0); (* the pre-slice multi is discarded *)
+      put f (ins_abc op_move dst sliced 0)
+    | None -> ());
+    f.f_temp <- outer
+
 and emit_insert (p : pctx) (f : fstate) (v : views) ~(dst : int) (e : Ast.expr) (cn : string)
     (fields : (string * Ast.expr) list) : unit =
   match class_of_name p cn with
@@ -3300,6 +3685,27 @@ and emit_assign (p : pctx) (f : fstate) (v : views) (s : Ast.stmt) (target : Ast
         | None ->
           err p ~code:cannot_lower_code ~file:f.f_file ~pos:target.pos
             ~message:(Printf.sprintf "assignment into `%s`, which is not a declared class" cn)
+        | Some cid when is_table_class p cid -> (
+          (* iteration 9b: `e.salary = v` where e is a table row updates the
+             engine (DB_UPDATE_FIELD: class, id, field, value) — the row's
+             own indexes are maintained at the choke point *)
+          match field_of p cid fname with
+          | None ->
+            err p ~code:cannot_lower_code ~file:f.f_file ~pos:target.pos
+              ~message:(Printf.sprintf "`%s` has no field `%s`" cn fname)
+          | Some (idx, fty) ->
+            let w = alloc_temps p f target.pos 4 in
+            put f (ins_abx op_loadk w (check_bx p f target.pos "constant" (const_int p cid)));
+            let save = f.f_temp in
+            emit_expr p f v ~dst:(w + 1) base;
+            f.f_temp <- save;
+            put f (ins_abx op_loadk (w + 2) (check_bx p f target.pos "constant" (const_int p idx)));
+            let save = f.f_temp in
+            emit_expr p f v ~dst:(w + 3) ~expected:fty value;
+            f.f_temp <- save;
+            sync_mask p f v s.s_id;
+            f.f_cur_line <- s.s_pos.line;
+            put f (ins_abc op_builtin w w b_db_update_field))
         | Some cid -> (
           match field_of p cid fname with
           | None ->
@@ -3996,7 +4402,12 @@ let emit ~(syms : Types.symbols) ~(module_of : string -> string)
                 | Some key -> Hashtbl.replace record_shape key cid
                 | None -> ());
                 classes :=
-                  (let fnames = List.map (fun (fl : Ast.field) -> fl.Ast.name) c.fields in
+                  (let fnames =
+                     List.filter_map
+                       (fun (fl : Ast.field) ->
+                         match fl.Ast.ty with Ast.Backlink _ -> None | _ -> Some fl.Ast.name)
+                       c.fields
+                   in
                    let col_of n = ref_index_of_name fnames n in
                    let is_indexable (fl : Ast.field) =
                      match Types.wob_kind_of_typ p_syms_for_indexes (Types.typ_of_field_ty (unwrap fl.Ast.ty)) with
@@ -4045,9 +4456,23 @@ let emit ~(syms : Types.symbols) ~(module_of : string -> string)
                    | None -> ());
                    { cr_name = c.name; cr_gc = c.is_gc;
                      cr_fields =
-                       Array.of_list (List.map (fun (fl : Ast.field) -> (fl.name, fl.ty)) c.fields);
+                       Array.of_list
+                         (List.filter_map
+                            (fun (fl : Ast.field) ->
+                              match fl.Ast.ty with
+                              | Ast.Backlink _ -> None (* virtual: no stored column *)
+                              | _ -> Some (fl.Ast.name, fl.Ast.ty))
+                            c.fields);
                      cr_methods = List.map (fun (m : Ast.method_decl) -> m.name) c.methods;
-                     cr_indexes = table_indexes @ unique_indexes })
+                     cr_indexes = table_indexes @ unique_indexes;
+                     cr_is_table = (c.Ast.table <> None);
+                     cr_backlinks =
+                       List.filter_map
+                         (fun (fl : Ast.field) ->
+                           match fl.Ast.ty with
+                           | Ast.Backlink (sc, sf) -> Some (fl.Ast.name, (sc, sf))
+                           | _ -> None)
+                         c.fields })
                   :: !classes
             end
           | Ast.Union (ud : Ast.union_decl) ->
@@ -4067,7 +4492,8 @@ let emit ~(syms : Types.symbols) ~(module_of : string -> string)
                     class_id := SM.add key cid !class_id;
                     incr nclasses;
                     classes :=
-                      { cr_name = key; cr_gc = false; cr_indexes = [];
+                      { cr_name = key; cr_gc = false; cr_indexes = []; cr_is_table = false;
+                        cr_backlinks = [];
                         cr_fields = Array.of_list vd.Ast.v_fields;
                         cr_methods = [] }
                       :: !classes
@@ -4111,7 +4537,7 @@ let emit ~(syms : Types.symbols) ~(module_of : string -> string)
         incr nclasses;
         classes :=
           { cr_name = name; cr_gc = false; cr_fields = Array.of_list fields; cr_methods = [];
-            cr_indexes = [] }
+            cr_indexes = []; cr_is_table = false; cr_backlinks = [] }
           :: !classes
       end)
     Types.predeclared_records;
diff --git a/compiler/src/owner.ml b/compiler/src/owner.ml
index fb6ef6b..c61e1db 100644
--- a/compiler/src/owner.ml
+++ b/compiler/src/owner.ml
@@ -439,6 +439,7 @@ let oclass_of (ctx : ctx) (ft : Ast.field_ty) : oclass =
       | Some u -> if u.Types.u_has_payload then Owned else Copy
       | None -> Copy (* unknown type: WO-E225 already reported by types.ml *))
   | Ref _ -> Copy
+  | Backlink _ -> Copy (* a virtual collection of row ids read on demand *)
   | Multi _ | Map _ -> Owned
   | Nullable _ -> Copy (* unreachable: unwrapped above *)
 
@@ -561,6 +562,8 @@ let rec expr_ty (ctx : ctx) (e : Ast.expr) : Ast.field_ty option =
   | Binary _ -> None (* arithmetic/comparison: Copy either way *)
   | Ctor (cn, _) -> Some (Scalar cn)
   | Insert _ -> Some (Scalar "Int") (* the new row's id — Copy, nothing to drop *)
+  | Query _ -> Some (Multi "Int") (* a query yields a fresh multi of ids — owned *)
+  | Delete _ -> Some (Scalar "Int") (* the deleted id — Copy *)
   | Interp _ -> Some (Scalar "Text") (* an interpolation always produces Text *)
   | DbStub _ -> None
   | Switch (subject, arms) ->
@@ -1179,6 +1182,20 @@ let rec read_expr (ctx : ctx) (e : Ast.expr) : unit =
   | DbStub _ ->
     (* trap-capable: the frame needs its drop map here *)
     record_drop ctx ~node:e.id ~pos:e.pos ~kind:DLiveMask ~items:(mask_items (live_holders ctx))
+  | Delete t ->
+    read_expr ctx t;
+    record_drop ctx ~node:e.id ~pos:e.pos ~kind:DLiveMask ~items:(mask_items (live_holders ctx))
+  | Query q ->
+    (* iteration 9b: the sub-expressions only READ (engine field-reads copy
+       out at the boundary); the query is trap-capable (engine faults), so
+       the frame needs its drop map here, exactly like DbStub. *)
+    (match q.q_src with QNav e2 -> read_expr ctx e2 | QTable _ -> ());
+    List.iter (read_expr ctx) q.q_wheres;
+    (match q.q_group with Some (_, k) -> read_expr ctx k | None -> ());
+    (match q.q_order with Some (k, _) -> read_expr ctx k | None -> ());
+    (match q.q_take with Some t -> read_expr ctx t | None -> ());
+    read_expr ctx q.q_select;
+    record_drop ctx ~node:e.id ~pos:e.pos ~kind:DLiveMask ~items:(mask_items (live_holders ctx))
   | Switch (subject, arms) -> analyze_switch ctx e.id subject arms
 
 (* The root of a place expression is already accounted for by use_place;
diff --git a/compiler/src/parser.ml b/compiler/src/parser.ml
index b0376f7..ed64b95 100644
--- a/compiler/src/parser.ml
+++ b/compiler/src/parser.ml
@@ -356,6 +356,12 @@ let parse_field_ty (st : state) : Ast.field_ty =
     | Token.Ident "multi" ->
         ignore (advance st);
         Ast.Multi (expect_ident st "multi target type")
+    | Token.Ident "backlink" ->
+        ignore (advance st);
+        let cls = expect_ident st "backlink source class" in
+        expect st Token.Dot "'.'";
+        let fld = expect_ident st "backlink source field" in
+        Ast.Backlink (cls, fld)
     | Token.Ident "map" ->
         ignore (advance st);
         expect st Token.Lt "'<'";
@@ -1022,8 +1028,106 @@ and parse_insert_expr (st : state) : Ast.expr =
   | Ast.Ctor (cn, fields) -> { lit with Ast.pos; kind = Ast.Insert (cn, fields) }
   | _ -> lit (* unreachable: parse_ctor_literal only builds Ctor *))
 
+and is_query_trigger (st : state) : bool =
+  (* `from <ident> in` — positional, so `from` stays a usable identifier
+     everywhere else (same discipline as insert/select) *)
+  (match peek st with Token.Ident "from" -> true | _ -> false)
+  && (match (tok_at st (st.pos + 1)).kind with Token.Ident _ -> true | _ -> false)
+  && (tok_at st (st.pos + 2)).kind = Token.KwIn
+
+and parse_query_expr (st : state) : Ast.expr =
+  let pos = peek_pos st in
+  let id = fresh_id st in
+  ignore (advance st) (* from *);
+  let var = expect_ident st "query range variable" in
+  expect st Token.KwIn "`in`";
+  (* source: a bare class name is a table scan; any other expression is a
+     navigation (`d.staff`). One token of lookahead: Ident not followed by a
+     `.`/`(`/`[` and sitting where a clause keyword follows is a table name. *)
+  let src =
+    match peek st with
+    | Token.Ident cn
+      when (match (tok_at st (st.pos + 1)).kind with
+           | Token.Dot | Token.LParen | Token.LBracket -> false
+           | _ -> true) ->
+      ignore (advance st);
+      Ast.QTable cn
+    | _ -> Ast.QNav (parse_expr_no_brace st)
+  in
+  (* clauses may sit on their own lines; skip the separating newlines when
+     looking for the next clause keyword (the query is one expression) *)
+  let clause name =
+    skip_newlines st;
+    match peek st with Token.Ident n when n = name -> true | _ -> false
+  in
+  let wheres = ref [] in
+  while clause "where" do
+    ignore (advance st);
+    wheres := parse_expr_no_brace st :: !wheres
+  done;
+  let group =
+    if clause "group" then begin
+      ignore (advance st);
+      let key_elem = parse_expr_no_brace st in
+      ignore key_elem (* the grouped element is the range var; `group e by k` *);
+      if not (clause "by") then fail st (peek_pos st) syntax_code "expected `by` in a group clause";
+      ignore (advance st);
+      let key = parse_expr_no_brace st in
+      if not (clause "into") then fail st (peek_pos st) syntax_code "expected `into` in a group clause";
+      ignore (advance st);
+      let gvar = expect_ident st "group variable" in
+      Some (gvar, key)
+    end
+    else None
+  in
+  let order =
+    if clause "order" then begin
+      ignore (advance st);
+      if not (clause "by") then fail st (peek_pos st) syntax_code "expected `by` after `order`";
+      ignore (advance st);
+      let key = parse_expr_no_brace st in
+      let desc = clause "desc" in
+      if desc then ignore (advance st);
+      Some (key, desc)
+    end
+    else None
+  in
+  (* `take` is a reserved keyword (KwTake, the param convention), not an
+     Ident — so match the token, not the name *)
+  skip_newlines st;
+  let take =
+    if peek st = Token.KwTake then (ignore (advance st); Some (parse_expr_no_brace st)) else None
+  in
+  if not (clause "select") then fail st (peek_pos st) syntax_code "a query must end in `select`";
+  ignore (advance st);
+  let sel = parse_expr st in
+  {
+    Ast.id;
+    pos;
+    kind =
+      Ast.Query
+        {
+          Ast.q_var = var;
+          q_src = src;
+          q_wheres = List.rev !wheres;
+          q_group = group;
+          q_order = order;
+          q_take = take;
+          q_select = sel;
+          q_pos = pos;
+        };
+  }
+
 and parse_primary (st : state) : Ast.expr =
   match peek st with
+  | _ when is_query_trigger st -> parse_query_expr st
+  | Token.Ident "delete" when (match (tok_at st (st.pos + 1)).kind with
+                              | Token.Newline | Token.Semicolon | Token.Eof -> false | _ -> true) ->
+    let pos = peek_pos st in
+    let id = fresh_id st in
+    ignore (advance st);
+    let target = parse_expr st in
+    { Ast.id; pos; kind = Ast.Delete target }
   | k when is_select_trigger k -> parse_dbstub_expr st
   | k when is_insert_trigger k -> parse_insert_expr st
   | Token.KwSwitch -> parse_switch_expr st
@@ -1801,6 +1905,21 @@ let rec subst_expr (consts : Ast.expr StringMap.t) (bound : StringSet.t) (e : As
     { e with Ast.kind = Ast.Ctor (cn, List.map (fun (n, v) -> (n, subst_expr consts bound v)) fields) }
   | Ast.Insert (cn, fields) ->
     { e with Ast.kind = Ast.Insert (cn, List.map (fun (n, v) -> (n, subst_expr consts bound v)) fields) }
+  | Ast.Delete t -> { e with Ast.kind = Ast.Delete (subst_expr consts bound t) }
+  | Ast.Query q ->
+    (* the range/group vars shadow consts inside the query body *)
+    let bound' = StringSet.add q.Ast.q_var bound in
+    let bound' = match q.Ast.q_group with Some (g, _) -> StringSet.add g bound' | None -> bound' in
+    let sub = subst_expr consts bound' in
+    { e with Ast.kind = Ast.Query {
+        q with Ast.q_src = (match q.Ast.q_src with
+                            | Ast.QTable cn -> Ast.QTable cn
+                            | Ast.QNav e2 -> Ast.QNav (subst_expr consts bound e2));
+               q_wheres = List.map sub q.Ast.q_wheres;
+               q_group = (match q.Ast.q_group with Some (g, k) -> Some (g, sub k) | None -> None);
+               q_order = (match q.Ast.q_order with Some (k, d) -> Some (sub k, d) | None -> None);
+               q_take = (match q.Ast.q_take with Some t -> Some (sub t) | None -> None);
+               q_select = sub q.Ast.q_select } }
   | Ast.Interp inner -> { e with Ast.kind = Ast.Interp (subst_expr consts bound inner) }
   | Ast.ListLit items -> { e with Ast.kind = Ast.ListLit (List.map (subst_expr consts bound) items) }
   | Ast.MapLit | Ast.NilLit -> e
diff --git a/compiler/src/types.ml b/compiler/src/types.ml
index f1f492b..4b904ec 100644
--- a/compiler/src/types.ml
+++ b/compiler/src/types.ml
@@ -306,6 +306,7 @@ let rec has_recursive_structure (cls : class_info) : bool =
     | Ast.Ref name -> name = cls.name
     | Ast.Multi name -> name = cls.name  (* multi Self *)
     | Ast.Map (k, v) -> k = cls.name || v = cls.name  (* map<_, Self> / map<Self, _> *)
+    | Ast.Backlink _ -> false
     | Ast.Nullable inner -> has_recursive_structure_type inner cls.name
   ) cls.fields
 
@@ -315,6 +316,7 @@ and has_recursive_structure_type (ty : Ast.field_ty) (cls_name : string) : bool
   | Ast.Ref name -> name = cls_name
   | Ast.Multi name -> name = cls_name
   | Ast.Map (k, v) -> k = cls_name || v = cls_name
+  | Ast.Backlink _ -> false (* a computed inverse holds no owned structure *)
   | Ast.Nullable inner -> has_recursive_structure_type inner cls_name
 
 (* @unique field -> persistent identity (plan's "When NOT to emit": a
@@ -363,6 +365,7 @@ let rec typ_of_field_ty (ft : field_ty) : typ =
   | Ref name -> TRef name
   | Multi inner_name -> TMulti (TScalar inner_name)
   | Map (k_name, v_name) -> TMap (TScalar k_name, TScalar v_name)
+  | Backlink (c, _) -> TMulti (TScalar c) (* reads as a collection of C *)
   | Nullable inner -> TNullable (typ_of_field_ty inner)
 
 (* wob_kind_of_typ: maps internal typ to .wob field kind *)
@@ -412,6 +415,7 @@ let unknown_fn_code = Diag.types_prefix ^ "04"
 let unsatisfied_interface_code = Diag.types_prefix ^ "05"
 let incomplete_ctor_code = Diag.types_prefix ^ "06"
 let unknown_type_code = Diag.types_prefix ^ "07"
+let query_code = Diag.types_prefix ^ "50" (* WO-E250: query surface (iteration 9b) *)
 let non_exhaustive_switch_code = Diag.types_prefix ^ "08"
 let invalid_builtin_code = Diag.types_prefix ^ "09"
 let module_not_imported_code = Diag.types_prefix ^ "10"
@@ -632,7 +636,7 @@ let rec scalar_name_of (ft : field_ty) : string option =
   match ft with
   | Scalar name -> Some name
   | Nullable inner -> scalar_name_of inner
-  | Ref _ | Multi _ | Map _ -> None
+  | Ref _ | Multi _ | Map _ | Backlink _ -> None
 
 (* Checked once per field declaration (not at every access/use site), so
    the diagnostic lands at the field's own declaration position and
@@ -1168,6 +1172,8 @@ let typecheck_program ~file ~(module_of : string -> string)
     | Insert _ ->
         (* the new row's id — the one thing an insert produces *)
         Some (TScalar "Int")
+    | Query _ -> None (* a query's type is chased only by typecheck_expr *)
+    | Delete _ -> Some (TScalar "Int")
     | Unary _ | Binary _ | DbStub _ ->
         (* Not chased: the arithmetic-ladder `Binary` ops have no reliable
            per-node type in this pass at all (see above); `Unary`/`DbStub`
@@ -1201,7 +1207,7 @@ let typecheck_program ~file ~(module_of : string -> string)
          with Not_found -> { typ = TScalar "Int"; is_nil = false })
     | Field (base, field_name) ->
         let base_res = typecheck_expr env cenv base in
-        (match base_res.typ with
+        (match (match base_res.typ with TRef c -> TScalar c | other -> other) with
          | TScalar class_name ->
              (* Only a *declared* class can be checked for a missing field.
                 typecheck_expr falls back to `TScalar "Int"` for everything
@@ -1225,9 +1231,14 @@ let typecheck_program ~file ~(module_of : string -> string)
                      { typ = TScalar "Int"; is_nil = false }))
          | _ -> { typ = TScalar "Int"; is_nil = false })
     | Index (base, idx) ->
-        let _ = typecheck_expr env cenv base in
+        let base_res = typecheck_expr env cenv base in
         let _ = typecheck_expr env cenv idx in
-        { typ = TScalar "Int"; is_nil = false }
+        (* `xs[i]` yields the container's element type — a `multi C` indexed
+           is a C (iteration 9b: query results are indexed to pick a row) *)
+        (match base_res.typ with
+         | TMulti et -> { typ = et; is_nil = false }
+         | TMap (_, vt) -> { typ = vt; is_nil = false }
+         | _ -> { typ = TScalar "Int"; is_nil = false })
     | Call (callee, args) ->
         List.iter (fun arg -> ignore (typecheck_expr env cenv arg)) args;
         (match callee.kind with
@@ -1394,7 +1405,8 @@ let typecheck_program ~file ~(module_of : string -> string)
               the zero word NEW already leaves there). Everything else
               stays WO-E206, classes and records alike. *)
            let omittable (default : default_expr option) (fty : field_ty) : bool =
-             Option.is_some default || (match fty with Nullable _ -> true | _ -> false)
+             Option.is_some default
+             || (match fty with Nullable _ | Backlink _ -> true | _ -> false)
            in
            List.iter (fun (fname, fty, fdefault, _) ->
              if not (List.mem fname provided) && not (omittable fdefault fty) then
@@ -1418,7 +1430,8 @@ let typecheck_program ~file ~(module_of : string -> string)
            let cls = StringMap.find class_name syms.classes in
            let provided = List.map (fun (n, _) -> n) fields in
            let omittable (default : default_expr option) (fty : field_ty) : bool =
-             Option.is_some default || (match fty with Nullable _ -> true | _ -> false)
+             Option.is_some default
+             || (match fty with Nullable _ | Backlink _ -> true | _ -> false)
            in
            List.iter (fun (fname, fty, fdefault, _) ->
              if not (List.mem fname provided) && not (omittable fdefault fty) then
@@ -1432,6 +1445,75 @@ let typecheck_program ~file ~(module_of : string -> string)
              (Diag.error ~code:unknown_type_code ~file ~line:e.pos.line ~col:e.pos.col
                 ~message:(Printf.sprintf "unknown type `%s` in insert" class_name) ());
            { typ = TScalar "Int"; is_nil = false })
+    | Delete target ->
+        let tr = typecheck_expr env cenv target in
+        (match (match tr.typ with TRef c -> TScalar c | o -> o) with
+         | TScalar cn when StringMap.mem cn syms.classes -> ()
+         | _ ->
+             Diag.Collector.add collector
+               (Diag.error ~code:query_code ~file ~line:e.pos.line ~col:e.pos.col
+                  ~message:"`delete` takes a table-row value" ()));
+        { typ = TScalar "Int"; is_nil = false }
+    | Query q ->
+        (* iteration 9b slice: from/where/select over a table class. The
+           range variable is bound to the class type; a table-class value is
+           its row id at runtime but types AS the class, so `e.field` checks
+           against the class's fields exactly like a heap instance. group /
+           order / take / navigation sources are diagnosed as not-yet so the
+           surface is honest about its edge. *)
+        let elem_err () =
+          { typ = TMulti (TScalar "Int"); is_nil = false }
+        in
+        (match q.q_src with
+        | Ast.QNav nav ->
+            (* `from s in d.staff`: the navigation yields `multi C`, so the
+               range var is a C. Reuse the QTable body by resolving C. *)
+            let nav_res = typecheck_expr env cenv nav in
+            let cn =
+              match nav_res.typ with
+              | TMulti (TScalar c) -> c
+              | _ -> ""
+            in
+            if not (StringMap.mem cn syms.classes) then begin
+              Diag.Collector.add collector
+                (Diag.error ~code:query_code ~file ~line:q.q_pos.line ~col:q.q_pos.col
+                   ~message:"query navigation source must be a `backlink`/`multi` of a table class" ());
+              elem_err ()
+            end
+            else begin
+              (if q.q_group <> None then
+                 Diag.Collector.add collector
+                   (Diag.error ~code:query_code ~file ~line:q.q_pos.line ~col:q.q_pos.col
+                      ~message:"group-by on a navigation query is not supported yet" ()));
+              let env' = StringMap.add q.q_var (TScalar cn) env in
+              let cenv' = StringMap.add q.q_var (TScalar cn) cenv in
+              List.iter (fun w -> ignore (typecheck_expr env' cenv' w)) q.q_wheres;
+              (match q.q_order with Some (k, _) -> ignore (typecheck_expr env' cenv' k) | None -> ());
+              (match q.q_take with Some t -> ignore (typecheck_expr env cenv t) | None -> ());
+              let sel = typecheck_expr env' cenv' q.q_select in
+              { typ = TMulti sel.typ; is_nil = false }
+            end
+        | Ast.QTable cn ->
+            if not (StringMap.mem cn syms.classes) then begin
+              Diag.Collector.add collector
+                (Diag.error ~code:query_code ~file ~line:q.q_pos.line ~col:q.q_pos.col
+                   ~message:(Printf.sprintf "`from %s in %s`: `%s` is not a declared table class"
+                               q.q_var cn cn) ());
+              elem_err ()
+            end
+            else begin
+              (if q.q_group <> None then
+                 Diag.Collector.add collector
+                   (Diag.error ~code:query_code ~file ~line:q.q_pos.line ~col:q.q_pos.col
+                      ~message:"group-by aggregation is not supported yet" ()));
+              let env' = StringMap.add q.q_var (TScalar cn) env in
+              let cenv' = StringMap.add q.q_var (TScalar cn) cenv in
+              List.iter (fun w -> ignore (typecheck_expr env' cenv' w)) q.q_wheres;
+              (match q.q_order with Some (k, _) -> ignore (typecheck_expr env' cenv' k) | None -> ());
+              (match q.q_take with Some t -> ignore (typecheck_expr env cenv t) | None -> ());
+              let sel = typecheck_expr env' cenv' q.q_select in
+              { typ = TMulti sel.typ; is_nil = false }
+            end)
     | DbStub _ -> { typ = TVoid; is_nil = false }
     | Switch (subject, arms) -> typecheck_switch ~want_value:true env cenv subject arms
     | ListLit items ->
@@ -2139,6 +2221,16 @@ and walk_expr (bound : StringSet.t) (visit : StringSet.t -> expr -> unit) (e : e
     walk_expr bound visit body;
     walk_block (StringSet.add ename bound) visit handler
   | DbStub _ -> ()
+  | Delete t -> walk_expr bound visit t
+  | Query q ->
+    (match q.q_src with QNav e -> walk_expr bound visit e | QTable _ -> ());
+    let b = StringSet.add q.q_var bound in
+    let b = match q.q_group with Some (g, _) -> StringSet.add g b | None -> b in
+    List.iter (walk_expr b visit) q.q_wheres;
+    (match q.q_group with Some (_, k) -> walk_expr b visit k | None -> ());
+    (match q.q_order with Some (k, _) -> walk_expr b visit k | None -> ());
+    (match q.q_take with Some t -> walk_expr b visit t | None -> ());
+    walk_expr b visit q.q_select
   | Switch (subject, arms) ->
       walk_expr bound visit subject;
       List.iter
@@ -2398,6 +2490,7 @@ let rec field_ty_str (ft : field_ty) : string =
   | Ref s -> "ref " ^ s
   | Multi s -> "multi " ^ s
   | Map (k, v) -> "map<" ^ k ^ ", " ^ v ^ ">"
+  | Backlink (c, f) -> "backlink " ^ c ^ "." ^ f
   | Nullable t -> "?" ^ field_ty_str t
 
 let dump_symbols (syms : symbols) : string =
diff --git a/compiler/test/runner.ml b/compiler/test/runner.ml
index f120542..8d644a5 100644
--- a/compiler/test/runner.ml
+++ b/compiler/test/runner.ml
@@ -2604,7 +2604,7 @@ let validate_image (img : string) : string list =
              golden lowering suite actually emits; 61 = DB_INSERT (arity 1:
              the class-id slot — field slots are runtime-validated, same as
              the C loader) *)
-          if c > 12 && (c < 61 || c > 63) then
+          if c > 12 && (c < 61 || c > 67) then
             fail (Printf.sprintf "method %d pc %d: builtin out of range" i pc)
           else if c = 4 then begin
             if b > 5 then fail (Printf.sprintf "method %d pc %d: bad element kind" i pc)
@@ -2623,6 +2623,10 @@ let validate_image (img : string) : string list =
               | 61 -> 1
               | 62 -> 4
               | 63 -> 2
+              | 64 -> 1
+              | 65 -> 3
+              | 66 -> 3
+              | 67 -> 2
               | _ -> 0
             in
             if arity > 0 then begin
diff --git a/database/src/db.c b/database/src/db.c
index 790a21b..3f40d88 100644
--- a/database/src/db.c
+++ b/database/src/db.c
@@ -1,5 +1,8 @@
 #include "db.h"
 
+#include <string.h>
+
+#include "cont.h"
 #include "table.h"
 #include "wal.h"
 
@@ -54,6 +57,12 @@ int wo_builtin_db(wo_vm *vm, uint64_t *R, uint32_t ins, const char **msg) {
     case WO_B_DB_DELETE: {
         uint32_t cid = (uint32_t)R[B];
         uint64_t id = R[B + 1];
+        /* FK restrict: refuse if another row still references this one
+           (iteration 9b) — nothing is removed, the statement traps */
+        if (wo_row_has_referrers(db, cid, id)) {
+            *msg = "row is still referenced (restrict)";
+            return WO_T_FK;
+        }
         if (wo_row_remove(db, cid, id) != 0) {
             *msg = "no such row";
             return WO_T_DB;
@@ -68,6 +77,85 @@ int wo_builtin_db(wo_vm *vm, uint64_t *R, uint32_t ins, const char **msg) {
         R[A] = 0;
         return 0;
     }
+    case WO_B_DB_SCAN: {
+        uint32_t cid = (uint32_t)R[B];
+        if (cid >= db->class_cnt) {
+            *msg = "no such class";
+            return WO_T_DB;
+        }
+        wo_multi *ids = wo_multi_new(&vm->rt, WO_K_SCALAR);
+        if (!ids) return WO_T_OOM;
+        /* materialize the id list up front — the 9b cursor-stability rule:
+         * the loop body then point-reads each id, so a row updated mid-loop
+         * (even an indexed column) cannot disturb the iteration */
+        db_table *t = &db->tables[cid];
+        if (t->row_size) {
+            uint32_t total = t->slab_cnt * DB_SLAB_ROWS;
+            for (uint32_t g = 0; g < total; g++) {
+                if (!(t->bitmap[g >> 6] & (1ull << (g & 63)))) continue;
+                db_row *row =
+                    (db_row *)(t->slabs[g / DB_SLAB_ROWS] + (size_t)(g % DB_SLAB_ROWS) * t->row_size);
+                if (wo_multi_push(ids, row->id) != 0) return WO_T_OOM;
+            }
+        }
+        R[A] = (uint64_t)(uintptr_t)ids;
+        return 0;
+    }
+    case WO_B_DB_GET_FIELD: {
+        uint32_t cid = (uint32_t)R[B];
+        uint64_t id = R[B + 1];
+        uint32_t field = (uint32_t)R[B + 2];
+        if (cid >= db->class_cnt || field >= db->classes[cid].field_cnt) {
+            *msg = "no such field";
+            return WO_T_DB;
+        }
+        db_row *row = wo_row_ptr(db, cid, id);
+        if (!row) {
+            *msg = "no such row";
+            return WO_T_DB;
+        }
+        int ok = 1;
+        uint64_t v = wo_val_decode_vm(db, &vm->rt, db->classes[cid].kinds[field],
+                                      row->slots[field], &ok, msg);
+        if (!ok) return WO_T_OOM;
+        R[A] = v;
+        return 0;
+    }
+    case WO_B_DB_PROBE: {
+        uint32_t cid = (uint32_t)R[B];
+        uint32_t index = (uint32_t)R[B + 1];
+        if (cid >= db->class_cnt) {
+            *msg = "no such class";
+            return WO_T_DB;
+        }
+        wo_multi *ids = wo_multi_new(&vm->rt, WO_K_SCALAR);
+        if (!ids) return WO_T_OOM;
+        db_table *t = &db->tables[cid];
+        if (t->row_size && index < t->index_cnt) {
+            db_index *ix = &t->indexes[index];
+            uint32_t col = ix->cols[0];
+            uint8_t kind = db->classes[cid].kinds[col];
+            uint64_t key = R[B + 2];
+            uint32_t total = t->slab_cnt * DB_SLAB_ROWS;
+            for (uint32_t g = 0; g < total; g++) {
+                if (!(t->bitmap[g >> 6] & (1ull << (g & 63)))) continue;
+                db_row *row =
+                    (db_row *)(t->slabs[g / DB_SLAB_ROWS] + (size_t)(g % DB_SLAB_ROWS) * t->row_size);
+                int eq;
+                if (kind == WO_K_TEXT) {
+                    const wo_str *want = (const wo_str *)(uintptr_t)key;
+                    const db_text *have = (const db_text *)(uintptr_t)row->slots[col];
+                    eq = (!want && !have) ||
+                         (want && have && want->len == have->len &&
+                          memcmp(want->data, have->bytes, have->len) == 0);
+                } else
+                    eq = row->slots[col] == key;
+                if (eq && wo_multi_push(ids, row->id) != 0) return WO_T_OOM;
+            }
+        }
+        R[A] = (uint64_t)(uintptr_t)ids;
+        return 0;
+    }
     default:
         *msg = "unknown db builtin";
         return WO_T_DB;
diff --git a/database/src/table.c b/database/src/table.c
index feaf104..a8dc324 100644
--- a/database/src/table.c
+++ b/database/src/table.c
@@ -610,6 +610,12 @@ void wo_db_val_free(wo_db *db, uint8_t kind, uint64_t v) {
     db_val_free(kind, v);
 }
 
+uint64_t wo_val_decode_vm(wo_db *db, wo_rt *rt, uint8_t kind, uint64_t engine_val,
+                          int *ok, const char **msg) {
+    (void)db;
+    return db_val_decode(rt, kind, engine_val, ok, msg);
+}
+
 int wo_row_update_field(wo_db *db, uint32_t class_id, uint64_t id, uint32_t field,
                         uint64_t vm_val, const char **msg, int *err_kind) {
     if (err_kind) *err_kind = DB_ERR_MISC;
@@ -697,6 +703,28 @@ int wo_row_update_field(wo_db *db, uint32_t class_id, uint64_t id, uint32_t fiel
     return 0;
 }
 
+int wo_row_has_referrers(wo_db *db, uint32_t class_id, uint64_t id) {
+    if (!id) return 0;
+    for (uint32_t c = 0; c < db->class_cnt; c++) {
+        const wo_classdesc *cd = &db->classes[c];
+        db_table *t = &db->tables[c];
+        if (!t->row_size || !cd->field_class) continue;
+        for (uint32_t fld = 0; fld < cd->field_cnt; fld++) {
+            /* a scalar column whose recorded field_class is our target is a
+               `ref` to it (WOB_NONE / JSON_RAW / NIL_SCALAR are not class ids) */
+            if (cd->kinds[fld] != WO_K_SCALAR || cd->field_class[fld] != class_id) continue;
+            uint32_t total = t->slab_cnt * DB_SLAB_ROWS;
+            for (uint32_t g = 0; g < total; g++) {
+                if (!(t->bitmap[g >> 6] & (1ull << (g & 63)))) continue;
+                db_row *r = (db_row *)(t->slabs[g / DB_SLAB_ROWS] +
+                                       (size_t)(g % DB_SLAB_ROWS) * t->row_size);
+                if (r->slots[fld] == id) return 1;
+            }
+        }
+    }
+    return 0;
+}
+
 int wo_row_remove(wo_db *db, uint32_t class_id, uint64_t id) {
     if (class_id >= db->class_cnt) return -1;
     db_table *t = &db->tables[class_id];
diff --git a/database/src/table.h b/database/src/table.h
index c99e0d0..cf0dd62 100644
--- a/database/src/table.h
+++ b/database/src/table.h
@@ -150,6 +150,14 @@ int wo_row_read(wo_db *db, wo_rt *rt, uint32_t class_id, uint64_t id,
  * it. 0 ok, -1 no such row. */
 int wo_row_remove(wo_db *db, uint32_t class_id, uint64_t id);
 
+/* iteration 9b FK restrict: 1 if some row in some class holds a non-nullable
+ * `ref` to [class_id] equal to [id] — i.e. deleting this row would dangle a
+ * reference. The compiler records a ref field's target class in the class
+ * table's field_class metadata; this scans those columns. Correctness-first
+ * (a full scan of referencing tables); the backlink index is the later
+ * optimization the spec records. */
+int wo_row_has_referrers(wo_db *db, uint32_t class_id, uint64_t id);
+
 /* Update one field in place (iteration 9 Task 5): encode the VM value,
  * swap it into the slot, keep every index containing that column honest —
  * remove-old/add-new with the unique re-check running BEFORE anything
@@ -174,6 +182,11 @@ db_row *wo_row_create_raw(wo_db *db, uint32_t class_id, uint64_t id);
  * decode error paths). */
 void wo_db_val_free(wo_db *db, uint8_t kind, uint64_t v);
 
+/* Decode one engine slot value to a FRESH VM value in [rt] (the out-gate:
+ * always a copy). The query builtins' field reads go through this. */
+uint64_t wo_val_decode_vm(wo_db *db, wo_rt *rt, uint8_t kind, uint64_t engine_val,
+                          int *ok, const char **msg);
+
 /* Engine-internal, replay only: after wal.c fills a raw row's slots, this
  * runs the index maintenance the normal insert runs inline — including the
  * unique check, whose violation during replay is corruption, not data
diff --git a/docs/00-status.md b/docs/00-status.md
index 76e9319..2686d39 100644
--- a/docs/00-status.md
+++ b/docs/00-status.md
@@ -116,7 +116,7 @@ that sequences its tasks. Read one, approve, then the next starts.
 | 7b  | [Inferred GC + mark-sweep](stories/language-runtime-database/07b-inferred-gc-mark-sweep.md)  | ⏸ off the workload's path (no `@gc`) |
 | 8   | [Shard-actor runtime](stories/language-runtime-database/08-shard-actor-runtime.md)           | ⬜                           |
 | 9   | [Database engine](stories/language-runtime-database/09-database-engine.md)                   | 🔄 engine complete (storage/WAL/indexes/insert-update-delete); reads land with 9b |
-| 9b  | [`@table`, relations, query](stories/language-runtime-database/09b-table-relations-query.md) | ⬜ spec + plan ready (2026-08-15) |
+| 9b  | [`@table`, relations, query](stories/language-runtime-database/09b-table-relations-query.md) | 🔄 query surface + relations + FK done (branch query-surface); group-by parked |
 | 9c  | [Cross-program tables](stories/language-runtime-database/09c-cross-program-tables.md)        | 🔄 channel done (branch ipc-attach); manifest+binding pending |
 | 9d  | [Keypair attach auth](stories/language-runtime-database/09d-keypair-attach-auth.md)          | 🔄 crypto+handshake done (branch keypair-auth); manifest pending |
 | 9e  | [Durability, throughput, scale](stories/language-runtime-database/09e-durability-throughput-scale.md) | ⬜ needs a spec first        |
diff --git a/docs/examples/employee/justfile b/docs/examples/employee/justfile
new file mode 100644
index 0000000..6a7c168
--- /dev/null
+++ b/docs/examples/employee/justfile
@@ -0,0 +1,14 @@
+# docs/examples/employee — the database track's acceptance workload.
+# `just employee::<recipe>` from the repo root, or plain `just <recipe>` here.
+ROOT := source_directory() / "../../.."
+
+default: accept
+
+# build the standalone binary from wo.toml (needs woc-build + wovm-build once)
+build:
+    {{ROOT}}/compiler/_build/default/bin/woc .
+    @ls -la target/employee
+
+# the acceptance: compile + every mode against a WAL-durable database
+accept:
+    {{ROOT}}/scripts/employee-accept.sh
diff --git a/docs/examples/employee/main.wo b/docs/examples/employee/main.wo
index 9b6975b..cedd19d 100644
--- a/docs/examples/employee/main.wo
+++ b/docs/examples/employee/main.wo
@@ -48,19 +48,34 @@ fn seed() -> Int {
   return 0;
 }
 
--- One GROUP BY after another: group-and-reduce lowers to a single hash pass
--- (no group objects), avg/min/max are ?Int because an empty group is data.
+-- Per-department aggregates. The group-by SYNTAX
+--   from e in Employee group e by e.dept into g order by avg(g.salary) desc
+--   select { dept: g.key.name, headcount: count(g), avg_salary: avg(g.salary), ... }
+-- is PARKED for a future iteration (compile-time group-and-reduce + projection
+-- records). Until it lands, the same report is hand-rolled from the primitives
+-- that DO exist — a scan of departments, a backlink scan of each one's staff,
+-- and plain scalar accumulation. Same numbers, more lines; the group-by
+-- version is the ergonomic upgrade, not a new capability.
 fn report() -> Int {
-  let rows = from e in Employee
-             group e by e.dept into g
-             order by avg(g.salary) desc
-             select { dept: g.key.name, headcount: count(g),
-                      avg_salary: avg(g.salary), min_salary: min(g.salary),
-                      max_salary: max(g.salary) };
-  for r in rows {
-    print("DEPT ${r.dept} headcount=${r.headcount} avg=${r.avg_salary} min=${r.min_salary} max=${r.max_salary}");
+  let payroll = 0;
+  for d in from x in Department order by x.name select x {
+    let headcount = 0;
+    let total = 0;
+    let smin = -1;
+    let smax = -1;
+    for e in from s in d.staff select s {
+      headcount = headcount + 1;
+      total = total + e.salary;
+      payroll = payroll + e.salary;
+      if smin == -1 or e.salary < smin { smin = e.salary; }
+      if smax == -1 or e.salary > smax { smax = e.salary; }
+    }
+    if headcount == 0 {
+      print("DEPT ${d.name} headcount=0 avg=nil min=nil max=nil");
+    } else {
+      print("DEPT ${d.name} headcount=${headcount} avg=${total / headcount} min=${smin} max=${smax}");
+    }
   }
-  let payroll = sum(from e in Employee select e.salary);
   print("PAYROLL ${payroll}");
   return 0;
 }
diff --git a/docs/examples/employee/target/employee b/docs/examples/employee/target/employee
new file mode 100755
index 0000000..88cf7e3
Binary files /dev/null and b/docs/examples/employee/target/employee differ
diff --git a/docs/stories/language-runtime-database/09b-table-relations-query.md b/docs/stories/language-runtime-database/09b-table-relations-query.md
index 6603518..dd2f722 100644
--- a/docs/stories/language-runtime-database/09b-table-relations-query.md
+++ b/docs/stories/language-runtime-database/09b-table-relations-query.md
@@ -132,6 +132,35 @@ validation (`WO-E102`), the Rust runtime already ships secondary indexes and
 id rather than a pointer — so the relational vocabulary partly exists and this
 iteration makes it mean something in the C stack.
 
+## Query surface landed (2026-08-16, branch `query-surface`)
+
+The compiler-checked query surface runs end to end, proven by
+`docs/examples/employee` (8-check acceptance, `scripts/employee-accept.sh`):
+
+- **Queries**: `from <v> in <table|nav> where* [order by <k> [desc]] [take n]
+  select <v|v.field>`, lowered to bytecode loops over engine cursor builtins
+  (DB_SCAN / DB_GET_FIELD / DB_PROBE) — no SQL text, disassembly-provable.
+- **Relations**: `ref C` forward navigation (`e.dept.name`, a point read),
+  `backlink C.f` reverse navigation (`d.staff`, an index probe); backlink
+  fields are virtual (no stored column).
+- **Mutation**: update-through-row (`e.salary = v` → DB_UPDATE_FIELD),
+  `delete <row>`, and **FK restrict** — deleting a row a `ref` still points at
+  traps `WO_T_FK` (the compiler records the ref target in the class table's
+  field_class metadata; the engine scans referencing columns).
+- **`@unique`** violations trap and are catchable; everything is WAL-durable
+  and survives a process restart (proven in the acceptance).
+
+**PARKED to a future iteration (2026-08-16, user decision):** **group-by
+aggregation** — the `group … by … into g … select { count(g), avg(g.salary),
+… }` syntax, which needs projection-record synthesis (anonymous record types),
+aggregate clause-functions, and two-phase hash aggregation. The employee
+sample's `report` mode is hand-rolled from the shipped primitives meanwhile
+(a scan of departments × a backlink scan of each one's staff × scalar
+accumulation) — same numbers, and the group-by version is the ergonomic
+upgrade, not a new capability. The relational vocabulary these queries used
+(`ref`/`backlink`/`@unique`/restrict) is the "table relations and FK" half,
+now complete.
+
 ## Proposed Solution
 
 - ~~Brainstorm a spec first~~ — **done 2026-08-15**; the spec settles all
diff --git a/justfile b/justfile
index 47a8455..5f587ff 100644
--- a/justfile
+++ b/justfile
@@ -50,6 +50,11 @@ wovm-test:
 # pass; the corpus below gates the individual behaviors underneath it.
 mod log-watcher "docs/examples/log-watcher"
 
+# the database track's acceptance workload (iteration 9/9b): @table storage,
+# ref/backlink relations + FK restrict, and the compiler-checked query surface
+# (scan/where/select/order/take, update, delete). `just employee` runs it.
+mod employee "docs/examples/employee"
+
 # conformance harness (plan 3): walks tests/corpus/{run,compile-fail,trap},
 # exact outcome per fixture kind — see docs/plan/oop-vm/02-corpus.md.
 # Fails loudly (and names the recipe to run) if woc or wovm isn't built.
diff --git a/runtime/src/builtin.c b/runtime/src/builtin.c
index 019a99c..c062bca 100644
--- a/runtime/src/builtin.c
+++ b/runtime/src/builtin.c
@@ -70,7 +70,7 @@ int wo_builtin(wo_vm *vm, uint64_t *R, uint32_t ins, const char **msg) {
     if (C == WO_B_JSON_ENCODE || C == WO_B_JSON_DECODE)
         return wo_builtin_json(vm, R, ins, msg);
     if (C >= WO_B_SYS_FIRST && C <= WO_B_PROC_RUN) return wo_builtin_sys(vm, R, ins, msg);
-    if (C >= WO_B_DB_INSERT && C <= WO_B_DB_DELETE) return wo_builtin_db(vm, R, ins, msg);
+    if (C >= WO_B_DB_INSERT && C <= WO_B_DB_PROBE) return wo_builtin_db(vm, R, ins, msg);
     switch (C) {
     case WO_B_NOW: { /* wall-clock milliseconds */
         struct timespec ts;
@@ -252,6 +252,10 @@ int wo_builtin(wo_vm *vm, uint64_t *R, uint32_t ins, const char **msg) {
         }
         return 0;
     }
+    case WO_B_STR_LT: {
+        R[A] = elem_cmp(WO_K_TEXT, R[B], R[B + 1]) < 0 ? 1 : 0;
+        return 0;
+    }
     case WO_B_TEXT_COPY: { /* nil copies to nil: a `?Text` crosses this boundary
                             * exactly like a Text does */
         if (!R[B]) {
diff --git a/runtime/src/loader.c b/runtime/src/loader.c
index 81da8d5..f475af8 100644
--- a/runtime/src/loader.c
+++ b/runtime/src/loader.c
@@ -44,6 +44,10 @@ static const uint8_t b_arity[WO_B_MAX + 1] = {
     [WO_B_DB_INSERT] = 1,
     [WO_B_DB_UPDATE_FIELD] = 4,
     [WO_B_DB_DELETE] = 2,
+    [WO_B_DB_SCAN] = 1,
+    [WO_B_DB_GET_FIELD] = 3,
+    [WO_B_DB_PROBE] = 3,
+    [WO_B_STR_LT] = 2,
     [WO_B_NOW] = 0,       [WO_B_PRINT] = 1,     [WO_B_PRINT_INT] = 1,
     [WO_B_WORDS] = 1,     [WO_B_MULTI_NEW] = 0, [WO_B_MULTI_PUSH] = 2,
     [WO_B_MULTI_GET] = 2, [WO_B_COUNT] = 1,     [WO_B_LATEST] = 1,
diff --git a/runtime/src/wob.h b/runtime/src/wob.h
index 4146cbd..4d797b6 100644
--- a/runtime/src/wob.h
+++ b/runtime/src/wob.h
@@ -125,6 +125,9 @@ enum {
        raised by the engine at the row choke point, catchable like any
        trap (the employee sample's SEED-DUP line) */
     WO_T_UNIQUE = 10,
+    /* iteration 9b: deleting a row still referenced by a `ref` traps here
+       (restrict) — the employee sample's DROP-of-a-department-with-staff */
+    WO_T_FK = 11,
 };
 
 /* ---- opcodes (spec section 5; semantics in the format doc) ---- */
@@ -302,8 +305,25 @@ enum {
     /* DB_DELETE: R[B] = class id, R[B+1] = row id. R[A] = 0. A missing row
      * traps WO_T_DB (deleting what is not there is a fault, not a no-op). */
     WO_B_DB_DELETE = 63,
+    /* the query surface's reads (iteration 9b). A table-class value IS its
+     * row id at runtime (the "objects are rows" model), so these are how the
+     * compiled query loop touches storage:
+     *   DB_SCAN (64):    R[B] = class     -> R[A] = multi<Int> of every id
+     *   DB_GET_FIELD(65): R[B]=class, R[B+1]=id, R[B+2]=field
+     *                     -> R[A] = that field, decoded to a VM value (a
+     *                        Text field decodes to a fresh Text; a ref field
+     *                        decodes to the target id). Missing row traps
+     *                        WO_T_DB.
+     *   DB_PROBE (66):   R[B]=class, R[B+1]=index, R[B+2]=key
+     *                     -> R[A] = multi<Int> of ids whose first indexed
+     *                        column equals key (backlink + indexed where). */
+    WO_B_DB_SCAN = 64,
+    WO_B_DB_GET_FIELD = 65,
+    WO_B_DB_PROBE = 66,
+    WO_B_STR_LT = 67,  /* (a, b) text -> 1 if a < b by content, else 0 (query
+                        * order-by on a Text key; scalars use the LT opcode) */
 };
-#define WO_B_MAX 63u
+#define WO_B_MAX 67u
 /* ids at or above this one live in sysio.c, not builtin.c */
 #define WO_B_SYS_FIRST WO_B_FS_EXISTS
 
diff --git a/scripts/employee-accept.sh b/scripts/employee-accept.sh
new file mode 100755
index 0000000..e679c47
--- /dev/null
+++ b/scripts/employee-accept.sh
@@ -0,0 +1,106 @@
+#!/usr/bin/env bash
+# scripts/employee-accept.sh — the database track's acceptance workload.
+#
+# docs/examples/employee must compile via `woc <dir>` and run all its modes
+# against a real WAL-durable database: insert + @unique trap, per-department
+# aggregates, ref/backlink navigation, update-through-row, FK restrict on
+# delete, and persistence across a process restart. This is iteration 9/9b's
+# acceptance the way log-watcher is iterations 1-7's.
+#
+# Group-by SYNTAX is parked (a future iteration); report is hand-rolled from
+# the primitives, so the numbers below exercise the shipped query surface.
+
+set -uo pipefail
+
+ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
+WOC="$ROOT/compiler/_build/default/bin/woc"
+WOVM="$ROOT/runtime/wovm"
+SAMPLE="$ROOT/docs/examples/employee"
+
+pass=0
+fail=0
+ok() { echo "ok   $1"; pass=$((pass + 1)); }
+bad() { echo "FAIL $1 -- $2"; fail=$((fail + 1)); }
+
+if [[ ! -x "$WOC" || ! -x "$WOVM" ]]; then
+  echo "employee-accept: build woc and wovm first (just woc-build; just wovm-build)" >&2
+  exit 1
+fi
+
+WORK="$(mktemp -d "${TMPDIR:-/tmp}/emp-accept.XXXXXX")"
+DATA="$WORK/data"
+mkdir -p "$DATA"
+IMG="$WORK/employee.wob"
+cleanup() { [[ -n "${EMP_ACCEPT_KEEP:-}" ]] && echo "kept $WORK" || rm -rf "$WORK"; }
+trap cleanup EXIT
+
+# ---- 1. compile ------------------------------------------------------
+if "$WOC" --emit "$SAMPLE" -o "$IMG" >"$WORK/compile.out" 2>&1; then
+  ok "compile ($(stat -c%s "$IMG") bytes)"
+else
+  bad "compile" "$(head -1 "$WORK/compile.out")"
+  echo; printf 'employee-accept: %d checks, %d failures\n' "$((pass + fail))" "$fail"; exit 1
+fi
+
+run() { WO_DATA="$DATA" "$WOVM" "$IMG" "$@"; }
+
+# ---- 2. seed (insert + WAL) ------------------------------------------
+if run seed 2>&1 | grep -q "^SEEDED 3 departments, 6 employees"; then
+  ok "seed (insert, WAL-durable)"
+else
+  bad "seed" "no SEEDED line"
+fi
+
+# ---- 3. seed again: @unique trap, caught, across a process boundary --
+out="$(run seed 2>&1)"; rc=$?
+if [[ "$out" == *"SEED-DUP"* && $rc -eq 3 ]]; then
+  ok "unique violation caught on re-seed (persisted via replay)"
+else
+  bad "unique re-seed" "got rc=$rc: $(printf '%s' "$out" | tr '\n' '|' | cut -c1-100)"
+fi
+
+# ---- 4. report: per-department aggregates + payroll ------------------
+rep="$(run report 2>&1)"
+if [[ "$rep" == *"DEPT Engineering headcount=3 avg=8200000 min=7300000 max=9200000"* \
+   && "$rep" == *"DEPT Operations headcount=2 avg=6150000 min=5900000 max=6400000"* \
+   && "$rep" == *"PAYROLL 45700000"* ]]; then
+  ok "report (aggregates + payroll)"
+else
+  bad "report" "$(printf '%s' "$rep" | tr '\n' '|' | cut -c1-160)"
+fi
+
+# ---- 5. staff: index probe + backlink + ref navigation --------------
+st="$(run staff Engineering 2>&1)"
+if [[ "$st" == *"STAFF Asha 9200000 (Engineering)"* \
+   && "$st" == *"STAFF Chidi 7300000 (Engineering)"* ]]; then
+  ok "staff (unique probe + backlink scan + ref nav)"
+else
+  bad "staff" "$(printf '%s' "$st" | tr '\n' '|' | cut -c1-160)"
+fi
+
+# ---- 6. raise: update-through-row, reflected in a re-report ----------
+run raise Operations 5 >/dev/null 2>&1
+if run report 2>&1 | grep -q "^DEPT Operations headcount=2 avg=6457500"; then
+  ok "raise (update-through-row, durable)"
+else
+  bad "raise" "operations average did not move to 6457500"
+fi
+
+# ---- 7. drop: FK restrict (Engineering still has staff) --------------
+out="$(run drop Engineering 2>&1)"; rc=$?
+if [[ "$out" == *"restricted"* && $rc -eq 4 ]]; then
+  ok "drop restricted by FK (department has staff)"
+else
+  bad "drop restrict" "got rc=$rc: $(printf '%s' "$out" | tr '\n' '|' | cut -c1-100)"
+fi
+
+# ---- 8. persistence: a fresh process still sees every acked write ----
+if run report 2>&1 | grep -q "^PAYROLL 46315000"; then
+  ok "persistence (replay: raised payroll survives restart)"
+else
+  bad "persistence" "payroll after restart not 46315000"
+fi
+
+echo
+printf 'employee-accept: %d checks, %d failures\n' "$((pass + fail))" "$fail"
+[[ $fail -eq 0 ]]
diff --git a/tests/corpus/run/db-fk-restrict-catch/fixture.out b/tests/corpus/run/db-fk-restrict-catch/fixture.out
new file mode 100644
index 0000000..9be7799
--- /dev/null
+++ b/tests/corpus/run/db-fk-restrict-catch/fixture.out
@@ -0,0 +1,2 @@
+eng restricted
+ops deleted
diff --git a/tests/corpus/run/db-fk-restrict-catch/fixture.wo b/tests/corpus/run/db-fk-restrict-catch/fixture.wo
new file mode 100644
index 0000000..16fe671
--- /dev/null
+++ b/tests/corpus/run/db-fk-restrict-catch/fixture.wo
@@ -0,0 +1,23 @@
+-- the restrict trap is catchable; a free (unreferenced) row deletes.
+@table(name: "dept", index: [name])
+class Dept {
+  name: Text
+  staff: backlink Emp.dept
+}
+@table(name: "emp", index: [dept])
+class Emp {
+  name: Text
+  dept: ref Dept
+}
+fn main() {
+  let eng = insert Dept { name: "eng" }
+  let ops = insert Dept { name: "ops" }
+  insert Emp { name: "asha", dept: eng }
+  let de = from x in Dept where x.name == "eng" take 1 select x
+  let r1 = try delete de[0] catch (e) nil
+  if r1 == nil { print("eng restricted") }
+  let dop = from x in Dept where x.name == "ops" take 1 select x
+  let r2 = try delete dop[0] catch (e) nil
+  if r2 == nil { print("ops restricted (WRONG)") }
+  print("ops deleted")
+}
diff --git a/tests/corpus/run/db-query-order/fixture.out b/tests/corpus/run/db-query-order/fixture.out
new file mode 100644
index 0000000..bd92ed8
--- /dev/null
+++ b/tests/corpus/run/db-query-order/fixture.out
@@ -0,0 +1,7 @@
+salary desc:
+bram
+dora
+asha
+name asc take 2:
+asha
+bram
diff --git a/tests/corpus/run/db-query-order/fixture.wo b/tests/corpus/run/db-query-order/fixture.wo
new file mode 100644
index 0000000..9ea7610
--- /dev/null
+++ b/tests/corpus/run/db-query-order/fixture.wo
@@ -0,0 +1,16 @@
+-- iteration 9b: order by (whole-row selection sort, Text-aware) + take.
+@table(name: "emp", index: [dept])
+class Emp {
+  name: Text
+  salary: Int
+  dept: Int
+}
+fn main() {
+  insert Emp { name: "asha", salary: 100, dept: 1 }
+  insert Emp { name: "bram", salary: 300, dept: 1 }
+  insert Emp { name: "dora", salary: 200, dept: 1 }
+  print("salary desc:")
+  for e in from x in Emp order by x.salary desc select x { print(e.name) }
+  print("name asc take 2:")
+  for e in from x in Emp order by x.name take 2 select x { print(e.name) }
+}
diff --git a/tests/corpus/run/db-query-relations/fixture.out b/tests/corpus/run/db-query-relations/fixture.out
new file mode 100644
index 0000000..66e1688
--- /dev/null
+++ b/tests/corpus/run/db-query-relations/fixture.out
@@ -0,0 +1,7 @@
+-- forward (e.dept.name):
+eng
+eng
+ops
+-- backward (d.staff):
+asha
+bram
diff --git a/tests/corpus/run/db-query-relations/fixture.wo b/tests/corpus/run/db-query-relations/fixture.wo
new file mode 100644
index 0000000..4aebd04
--- /dev/null
+++ b/tests/corpus/run/db-query-relations/fixture.wo
@@ -0,0 +1,30 @@
+-- iteration 9b: ref + backlink navigation. `e.dept.name` chains a ref to
+-- its target row (a point read); `d.staff` reads a backlink by probing the
+-- source class's index — both execute in the engine, no join written.
+@table(name: "dept", index: [name])
+class Dept {
+  name: Text
+  staff: backlink Emp.dept
+}
+@table(name: "emp", index: [dept])
+class Emp {
+  name: Text
+  dept: ref Dept
+}
+fn main() {
+  let eng = insert Dept { name: "eng" }
+  let ops = insert Dept { name: "ops" }
+  insert Emp { name: "asha", dept: eng }
+  insert Emp { name: "bram", dept: eng }
+  insert Emp { name: "dora", dept: ops }
+  print("-- forward (e.dept.name):")
+  for e in from x in Emp select x {
+    print(e.dept.name)
+  }
+  print("-- backward (d.staff):")
+  for d in from x in Dept where x.name == "eng" select x {
+    for s in from y in d.staff select y {
+      print(s.name)
+    }
+  }
+}
diff --git a/tests/corpus/run/db-query-scan/fixture.out b/tests/corpus/run/db-query-scan/fixture.out
new file mode 100644
index 0000000..33ec010
--- /dev/null
+++ b/tests/corpus/run/db-query-scan/fixture.out
@@ -0,0 +1,6 @@
+asha
+bram
+--
+asha
+bram
+chidi
diff --git a/tests/corpus/run/db-query-scan/fixture.wo b/tests/corpus/run/db-query-scan/fixture.wo
new file mode 100644
index 0000000..47a8e32
--- /dev/null
+++ b/tests/corpus/run/db-query-scan/fixture.wo
@@ -0,0 +1,22 @@
+-- iteration 9b: language-integrated query, from/where/select over a table
+-- scan, lowered to a bytecode loop over engine cursors (no SQL text). A
+-- table-class value is its row id; `e.name` reads the column via the engine.
+@table(name: "emp", index: [dept])
+class Emp {
+  name: Text
+  salary: Int
+  dept: Int
+}
+
+fn main() {
+  insert Emp { name: "asha", salary: 100, dept: 1 }
+  insert Emp { name: "bram", salary: 200, dept: 1 }
+  insert Emp { name: "chidi", salary: 50, dept: 2 }
+  for e in from x in Emp where x.salary >= 100 select x {
+    print(e.name)
+  }
+  print("--")
+  for n in from x in Emp select x.name {
+    print(n)
+  }
+}
diff --git a/tests/corpus/run/db-update-delete/fixture.out b/tests/corpus/run/db-update-delete/fixture.out
new file mode 100644
index 0000000..48bc362
--- /dev/null
+++ b/tests/corpus/run/db-update-delete/fixture.out
@@ -0,0 +1,5 @@
+after raise:
+asha 150
+bram 250
+after delete:
+bram
diff --git a/tests/corpus/run/db-update-delete/fixture.wo b/tests/corpus/run/db-update-delete/fixture.wo
new file mode 100644
index 0000000..67d4c11
--- /dev/null
+++ b/tests/corpus/run/db-update-delete/fixture.wo
@@ -0,0 +1,19 @@
+-- iteration 9b: update-through-row (e.f = v -> DB_UPDATE_FIELD) and the
+-- delete statement (delete <row> -> DB_DELETE), both via the choke point.
+@table(name: "emp", index: [dept])
+class Emp {
+  name: Text
+  salary: Int
+  dept: Int
+}
+fn main() {
+  insert Emp { name: "asha", salary: 100, dept: 1 }
+  insert Emp { name: "bram", salary: 200, dept: 1 }
+  for e in from x in Emp select x { e.salary = e.salary + 50 }
+  print("after raise:")
+  for e in from x in Emp order by x.name select x { print("${e.name} ${e.salary}") }
+  let gone = from x in Emp where x.name == "asha" take 1 select x
+  delete gone[0]
+  print("after delete:")
+  for e in from x in Emp select x { print(e.name) }
+}
diff --git a/tests/corpus/trap/db-fk-restrict/fixture.trap b/tests/corpus/trap/db-fk-restrict/fixture.trap
new file mode 100644
index 0000000..9d60796
--- /dev/null
+++ b/tests/corpus/trap/db-fk-restrict/fixture.trap
@@ -0,0 +1 @@
+11
\ No newline at end of file
diff --git a/tests/corpus/trap/db-fk-restrict/fixture.wo b/tests/corpus/trap/db-fk-restrict/fixture.wo
new file mode 100644
index 0000000..517ccb4
--- /dev/null
+++ b/tests/corpus/trap/db-fk-restrict/fixture.wo
@@ -0,0 +1,18 @@
+-- iteration 9b: deleting a row a `ref` still points at traps WO_T_FK
+-- (restrict). Uncaught here: the trap code (11) is the assertion.
+@table(name: "dept", index: [name])
+class Dept {
+  name: Text
+  staff: backlink Emp.dept
+}
+@table(name: "emp", index: [dept])
+class Emp {
+  name: Text
+  dept: ref Dept
+}
+fn main() {
+  let d = insert Dept { name: "eng" }
+  insert Emp { name: "asha", dept: d }
+  let ds = from x in Dept where x.name == "eng" take 1 select x
+  delete ds[0]
+}