From 71d34eb33cd79dd415dcfb7498d083f861b73801 Mon Sep 17 00:00:00 2001
From: Simone Carolini <simonecarolini.sc@gmail.com>
Date: Fri, 9 Oct 2026 13:18:47 +0200
Subject: [PATCH 1/5] feat(contract): mode_type and primary_key on Node,
 per-kind mode sets

MODE_TYPES and DEFAULT_MODE_TYPE (replace) join the model; Node carries
mode_type and primary_key as written. KindRules.mode_types allows every mode
for python-node and python-api and only replace for python-csv.

Signed-off-by: Simone Carolini <simonecarolini.sc@gmail.com>
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_0137E9XmyUXhwPTRrtNfivDw
---
 continuo_python_runtime/contract/kinds.py | 17 ++++++++----
 continuo_python_runtime/contract/model.py |  7 +++++
 tests/contract/test_kinds.py              | 17 +++++++++++-
 tests/contract/test_model.py              | 33 ++++++++++++++++++++++-
 4 files changed, 67 insertions(+), 7 deletions(-)

diff --git a/continuo_python_runtime/contract/kinds.py b/continuo_python_runtime/contract/kinds.py
index 926a6d1..3d7f758 100644
--- a/continuo_python_runtime/contract/kinds.py
+++ b/continuo_python_runtime/contract/kinds.py
@@ -1,8 +1,9 @@
 """Per-kind contract rules: which fields a node of each kind must carry.
 
 Each kind is one ``KindRules`` entry composed of a script policy, a reads rule,
-and whether it may name a Secret. Adding a kind means adding one entry to
-``RULES``; nothing that parses a contract branches on the kind's name.
+whether it may name a Secret, and which write modes it may declare. Adding a
+kind means adding one entry to ``RULES``; nothing that parses a contract
+branches on the kind's name.
 """
 
 from __future__ import annotations
@@ -14,7 +15,7 @@
 from continuo_engine_contract.sql import ensure_single_read  # type: ignore[import-untyped]
 from sqlglot.errors import TokenError
 
-from continuo_python_runtime.contract.model import KINDS
+from continuo_python_runtime.contract.model import DEFAULT_MODE_TYPE, KINDS, MODE_TYPES
 from continuo_python_runtime.csv_source import parse_csv_uri
 from continuo_python_runtime.errors import ContractError
 
@@ -108,12 +109,18 @@ class KindRules:
     script_required: bool
     reads: ReadsRule
     secret_allowed: bool = False
+    # The write modes a node of this kind may declare. A new kind gets replace
+    # only unless its entry opts into more.
+    mode_types: frozenset[str] = frozenset({DEFAULT_MODE_TYPE})
 
 
 RULES: dict[str, KindRules] = {
-    "python-node": KindRules(script_required=True, reads=SqlReads()),
+    "python-node": KindRules(script_required=True, reads=SqlReads(), mode_types=MODE_TYPES),
+    # A csv drop is a snapshot of its source: replace only (spec D2).
     "python-csv": KindRules(script_required=False, reads=CsvRead()),
-    "python-api": KindRules(script_required=True, reads=NoReads(), secret_allowed=True),
+    "python-api": KindRules(
+        script_required=True, reads=NoReads(), secret_allowed=True, mode_types=MODE_TYPES
+    ),
 }
 
 
diff --git a/continuo_python_runtime/contract/model.py b/continuo_python_runtime/contract/model.py
index 8be0fa2..bb9097b 100644
--- a/continuo_python_runtime/contract/model.py
+++ b/continuo_python_runtime/contract/model.py
@@ -8,6 +8,11 @@
 EXTRA_COLUMNS_POLICIES = frozenset({"raise", "warn"})
 KINDS = frozenset({"python-node", "python-csv", "python-api"})
 DEFAULT_KIND = "python-node"
+# How the harness writes a node's table on each run: replace empties it first,
+# append keeps the rows of earlier runs, upsert overwrites rows whose
+# primary_key exists and keeps the rest.
+MODE_TYPES = frozenset({"append", "upsert", "replace"})
+DEFAULT_MODE_TYPE = "replace"
 CONTRACT_VERSION = 1
 
 
@@ -38,6 +43,8 @@ class Node:
     content_hash: str | None = None
     kind: str = DEFAULT_KIND
     secret_ref: str = ""
+    mode_type: str = DEFAULT_MODE_TYPE
+    primary_key: tuple[str, ...] = ()
 
     @property
     def relation(self) -> str:
diff --git a/tests/contract/test_kinds.py b/tests/contract/test_kinds.py
index a011f87..1bf35bc 100644
--- a/tests/contract/test_kinds.py
+++ b/tests/contract/test_kinds.py
@@ -1,7 +1,7 @@
 import pytest
 
 from continuo_python_runtime.contract.kinds import RULES, CsvRead, NoReads, SqlReads, rules_for
-from continuo_python_runtime.contract.model import KINDS
+from continuo_python_runtime.contract.model import KINDS, MODE_TYPES
 from continuo_python_runtime.contract.secret_ref import validate_secret_ref
 from continuo_python_runtime.errors import ContractError
 
@@ -94,3 +94,18 @@ def test_secret_ref_accepts_valid_names(name):
 def test_secret_ref_rejects_invalid_names(name):
     with pytest.raises(ContractError, match="'secret_ref'"):
         validate_secret_ref(name, "L")
+
+
+def test_python_csv_only_replaces():
+    assert RULES["python-csv"].mode_types == frozenset({"replace"})
+
+
+@pytest.mark.parametrize("kind", ["python-node", "python-api"])
+def test_script_kinds_allow_every_mode(kind):
+    assert RULES[kind].mode_types == MODE_TYPES
+
+
+def test_every_kind_allows_replace_and_nothing_outside_the_vocabulary():
+    for rules in RULES.values():
+        assert "replace" in rules.mode_types
+        assert rules.mode_types <= MODE_TYPES
diff --git a/tests/contract/test_model.py b/tests/contract/test_model.py
index b690b72..13f24c8 100644
--- a/tests/contract/test_model.py
+++ b/tests/contract/test_model.py
@@ -1,4 +1,11 @@
-from continuo_python_runtime.contract.model import DEFAULT_KIND, KINDS, Column, Node
+from continuo_python_runtime.contract.model import (
+    DEFAULT_KIND,
+    DEFAULT_MODE_TYPE,
+    KINDS,
+    MODE_TYPES,
+    Column,
+    Node,
+)
 
 
 def _node(**over):
@@ -43,3 +50,27 @@ def test_kind_accepts_python_csv():
 
 def test_kinds_frozenset_contains_every_kind():
     assert KINDS == frozenset({"python-node", "python-csv", "python-api"})
+
+
+def test_mode_types_are_the_closed_vocabulary():
+    assert MODE_TYPES == frozenset({"append", "upsert", "replace"})
+    assert DEFAULT_MODE_TYPE == "replace"
+
+
+def test_mode_type_defaults_to_replace_with_no_key():
+    n = _node()
+    assert n.mode_type == "replace"
+    assert n.primary_key == ()
+
+
+def test_an_upsert_node_keeps_its_key_as_written_in_key_order():
+    n = _node(
+        output_columns=(
+            Column("rate_date", "DATE", nullable=False),
+            Column("base", "VARCHAR(3)", nullable=False),
+        ),
+        mode_type="upsert",
+        primary_key=("base", "rate_date"),
+    )
+    assert n.mode_type == "upsert"
+    assert n.primary_key == ("base", "rate_date")

From a9138cab1d8155da3b1678787814863029af6901 Mon Sep 17 00:00:00 2001
From: Simone Carolini <simonecarolini.sc@gmail.com>
Date: Fri, 9 Oct 2026 13:22:38 +0200
Subject: [PATCH 2/5] feat(contract): validate mode_type and primary_key in the
 loader

mode_type is one of append, upsert, replace (default replace) and must be
allowed for the node's kind. primary_key is required for upsert and rejected
otherwise; it is a non-empty list of distinct names (compared
case-insensitively), each naming a nullable: false output column exactly as
spelled there.

Signed-off-by: Simone Carolini <simonecarolini.sc@gmail.com>
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_0137E9XmyUXhwPTRrtNfivDw
---
 CHANGELOG.md                               |  10 ++
 continuo_python_runtime/contract/loader.py |  75 ++++++++++-
 tests/contract/test_loader.py              | 141 +++++++++++++++++++++
 3 files changed, 225 insertions(+), 1 deletion(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 3c21ffa..2c832ee 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -5,6 +5,16 @@ follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
 
 ## [Unreleased]
 
+### Added
+
+- Contracts may set `mode_type` — `replace` (the default), `append` or
+  `upsert` — and, for `upsert` only, `primary_key`: a non-empty list of output
+  columns, each spelled exactly as in `output_columns` and declared
+  `nullable: false`, names unique case-insensitively.
+  `python-csv` nodes allow only `replace`. `continuo-runtime validate`, `merge`
+  and `run` enforce these rules; the harness's write path for the new modes
+  follows in a later entry.
+
 ### Changed
 
 - **Breaking for newly copied templates.** The `template/` release workflow
diff --git a/continuo_python_runtime/contract/loader.py b/continuo_python_runtime/contract/loader.py
index 502adc2..c5f8927 100644
--- a/continuo_python_runtime/contract/loader.py
+++ b/continuo_python_runtime/contract/loader.py
@@ -17,7 +17,9 @@
 from continuo_python_runtime.contract.model import (
     CRITICALITIES,
     DEFAULT_KIND,
+    DEFAULT_MODE_TYPE,
     EXTRA_COLUMNS_POLICIES,
+    MODE_TYPES,
     Column,
     Node,
 )
@@ -40,6 +42,8 @@
     "config",
     "content_hash",
     "secret_ref",
+    "mode_type",
+    "primary_key",
 }
 
 _REQUIRED_STRING_FIELDS = ("schema", "table", "owner", "schedule")
@@ -140,6 +144,59 @@ def _check(value: Any, key: Any) -> None:
     return raw
 
 
+def _validate_primary_key(
+    raw: dict[str, Any], mode_type: str, columns: list[Column], label: str
+) -> tuple[str, ...]:
+    """Validate ``primary_key`` against ``mode_type`` and the declared columns.
+
+    Required for ``upsert`` and rejected otherwise: a key on ``append`` or
+    ``replace`` would look enforced when it is not. Names keep their order,
+    which is what the wire entry carries. Each must be spelled exactly as an
+    output column (postgres quotes identifiers, so ``Rate_Date`` is not
+    ``rate_date``) declared ``nullable: false``, because a NULL key never
+    matches an existing row. Repeats are refused case-insensitively too, so a
+    key cannot name two columns that differ only in case.
+    """
+    if mode_type != "upsert":
+        if "primary_key" in raw:
+            raise ContractError(
+                f"{label}: 'primary_key' is only allowed when mode_type is upsert "
+                f"(got {mode_type})"
+            )
+        return ()
+    if "primary_key" not in raw:
+        raise ContractError(f"{label}: 'primary_key' is required when mode_type is upsert")
+    raw_key = raw["primary_key"]
+    if not isinstance(raw_key, list) or not raw_key:
+        raise ContractError(
+            f"{label}: 'primary_key' must be a non-empty list of output column names"
+        )
+    by_name = {column.name: column for column in columns}
+    seen: set[str] = set()
+    for name in raw_key:
+        if not isinstance(name, str) or not name.strip():
+            raise ContractError(
+                f"{label}: 'primary_key' entry {name!r} must be a non-empty string"
+            )
+        folded = name.lower()  # same folding as continuo's parser
+        if folded in seen:
+            raise ContractError(
+                f"{label}: 'primary_key' names {name!r} more than once "
+                "(names are compared case-insensitively)"
+            )
+        seen.add(folded)
+        column = by_name.get(name)
+        if column is None:
+            raise ContractError(
+                f"{label}: 'primary_key' column {name!r} is not in output_columns"
+            )
+        if column.nullable:
+            raise ContractError(
+                f"{label}: 'primary_key' column {name!r} must be declared nullable: false"
+            )
+    return tuple(raw_key)
+
+
 def parse_node(
     raw: dict[str, Any],
     source: str,
@@ -158,7 +215,8 @@ def parse_node(
     ``check_reads=False`` skips *only* that :func:`ensure_single_read` call —
     the read-shape gate — leaving every other rule here (required fields,
     criticality, the ``reads`` map's own shape, output-column types and
-    uniqueness, ``config``, ``content_hash``) in force. It exists for
+    uniqueness, ``mode_type`` and ``primary_key``, ``config``,
+    ``content_hash``) in force. It exists for
     :func:`~continuo_python_runtime.harness.run_node`; see
     :func:`load_contract_dir` for why the runtime opts out.
     """
@@ -229,6 +287,17 @@ def parse_node(
             raise ContractError(f"{label}: 'secret_ref' is not allowed for kind {kind}")
         secret_ref = validate_secret_ref(raw["secret_ref"], label)
 
+    mode_type = raw.get("mode_type", DEFAULT_MODE_TYPE)
+    if not isinstance(mode_type, str) or mode_type not in MODE_TYPES:
+        raise ContractError(
+            f"{label}: 'mode_type' must be one of {sorted(MODE_TYPES)}, got {mode_type!r}"
+        )
+    if mode_type not in rules.mode_types:
+        raise ContractError(
+            f"{label}: 'mode_type' {mode_type} is not allowed for kind {kind} "
+            f"(allowed: {sorted(rules.mode_types)})"
+        )
+
     raw_columns = raw.get("output_columns")
     if not isinstance(raw_columns, list) or not raw_columns:
         raise ContractError(f"{label}: 'output_columns' must be a non-empty list")
@@ -269,6 +338,8 @@ def parse_node(
         seen_names.add(name)
         columns.append(Column(name=name, type=col_type, nullable=nullable))
 
+    primary_key = _validate_primary_key(raw, mode_type, columns, label)
+
     description = raw.get("description", "")
     if not isinstance(description, str):
         raise ContractError(f"{label}: 'description' must be a string")
@@ -294,6 +365,8 @@ def parse_node(
         content_hash=content_hash,
         kind=kind,
         secret_ref=secret_ref,
+        mode_type=mode_type,
+        primary_key=primary_key,
     )
 
 
diff --git a/tests/contract/test_loader.py b/tests/contract/test_loader.py
index 12c73c3..94475ef 100644
--- a/tests/contract/test_loader.py
+++ b/tests/contract/test_loader.py
@@ -727,3 +727,144 @@ def test_secret_ref_on_python_node_is_rejected():
 def test_bad_secret_ref_is_rejected_naming_the_node():
     with pytest.raises(ContractError, match=r"c\.yml \(analytics\.fx\): 'secret_ref'"):
         parse_node(_api_entry(secret_ref="continuo-app-credentials"), "c.yml")
+
+
+# --- write modes: mode_type and primary_key ----------------------------------
+
+
+def _upsert_entry(**over):
+    entry = {
+        "schema": "fx", "table": "fx_rates", "owner": "m", "schedule": "hourly",
+        "criticality": "SECONDARY", "kind": "python-api", "script": "scripts/fx.py",
+        "mode_type": "upsert", "primary_key": ["rate_date", "base"],
+        "output_columns": [
+            {"name": "rate_date", "type": "DATE", "nullable": False},
+            {"name": "base", "type": "VARCHAR(3)", "nullable": False},
+            {"name": "rate", "type": "DECIMAL(12,6)"},
+        ],
+    }
+    entry.update(over)
+    return entry
+
+
+def test_mode_type_defaults_to_replace():
+    node = parse_node(_node_entry(), "f.yml")
+    assert node.mode_type == "replace"
+    assert node.primary_key == ()
+
+
+@pytest.mark.parametrize("mode", ["append", "replace"])
+def test_python_node_accepts_append_and_explicit_replace(mode):
+    node = parse_node(_node_entry(mode_type=mode), "f.yml")
+    assert node.mode_type == mode
+    assert node.primary_key == ()
+
+
+def test_python_node_accepts_upsert_with_a_key():
+    node = parse_node(_node_entry(mode_type="upsert", primary_key=["id"]), "f.yml")
+    assert node.mode_type == "upsert"
+    assert node.primary_key == ("id",)
+
+
+def test_upsert_node_keeps_its_key_in_authored_order():
+    node = parse_node(_upsert_entry(primary_key=["base", "rate_date"]), "c.yml")
+    assert node.mode_type == "upsert"
+    assert node.primary_key == ("base", "rate_date")
+
+
+@pytest.mark.parametrize("mode", ["APPEND", "merge", "", None, 1, ["append"]])
+def test_unknown_mode_type_is_rejected(mode):
+    with pytest.raises(
+        ContractError,
+        match=r"c\.yml \(fx\.fx_rates\): 'mode_type' must be one of \['append', 'replace', 'upsert'\]",
+    ):
+        parse_node(_upsert_entry(mode_type=mode), "c.yml")
+
+
+@pytest.mark.parametrize("mode", ["append", "upsert"])
+def test_python_csv_only_allows_replace(mode):
+    entry = _csv_entry(mode_type=mode)
+    if mode == "upsert":
+        entry["primary_key"] = ["order_id"]
+    with pytest.raises(ContractError, match=f"'mode_type' {mode} is not allowed for kind python-csv"):
+        parse_node(entry, "f.yml", check_reads=False)
+
+
+def test_python_csv_accepts_explicit_replace():
+    node = parse_node(_csv_entry(mode_type="replace"), "f.yml", check_reads=False)
+    assert node.mode_type == "replace"
+
+
+def test_upsert_without_primary_key_is_rejected():
+    entry = _upsert_entry()
+    del entry["primary_key"]
+    with pytest.raises(ContractError, match="'primary_key' is required when mode_type is upsert"):
+        parse_node(entry, "c.yml")
+
+
+@pytest.mark.parametrize("mode", ["append", "replace"])
+def test_primary_key_without_upsert_is_rejected(mode):
+    with pytest.raises(
+        ContractError, match=rf"'primary_key' is only allowed when mode_type is upsert \(got {mode}\)"
+    ):
+        parse_node(_upsert_entry(mode_type=mode), "c.yml")
+
+
+def test_primary_key_on_the_default_mode_is_rejected():
+    entry = _upsert_entry()
+    del entry["mode_type"]
+    with pytest.raises(
+        ContractError, match=r"'primary_key' is only allowed when mode_type is upsert \(got replace\)"
+    ):
+        parse_node(entry, "c.yml")
+
+
+@pytest.mark.parametrize("key", [[], "rate_date", {"rate_date": 1}, None])
+def test_primary_key_must_be_a_non_empty_list(key):
+    with pytest.raises(
+        ContractError, match="'primary_key' must be a non-empty list of output column names"
+    ):
+        parse_node(_upsert_entry(primary_key=key), "c.yml")
+
+
+@pytest.mark.parametrize("name", ["", "  ", 3, None])
+def test_primary_key_entries_must_be_non_empty_strings(name):
+    with pytest.raises(ContractError, match="'primary_key' entry .* must be a non-empty string"):
+        parse_node(_upsert_entry(primary_key=["rate_date", name]), "c.yml")
+
+
+def test_primary_key_names_must_be_unique_case_insensitively():
+    with pytest.raises(ContractError, match="'primary_key' names 'RATE_DATE' more than once"):
+        parse_node(_upsert_entry(primary_key=["rate_date", "RATE_DATE"]), "c.yml")
+
+
+def test_primary_key_column_must_be_declared():
+    with pytest.raises(
+        ContractError,
+        match=r"c\.yml \(fx\.fx_rates\): 'primary_key' column 'quote' is not in output_columns",
+    ):
+        parse_node(_upsert_entry(primary_key=["rate_date", "quote"]), "c.yml")
+
+
+def test_primary_key_column_must_be_not_null():
+    with pytest.raises(
+        ContractError, match="'primary_key' column 'rate' must be declared nullable: false"
+    ):
+        parse_node(_upsert_entry(primary_key=["rate_date", "rate"]), "c.yml")
+
+
+def test_primary_key_names_must_match_output_columns_exactly():
+    """Review Focus 2: postgres quotes identifiers, so Rate_Date is not rate_date."""
+    with pytest.raises(
+        ContractError,
+        match=r"c\.yml \(fx\.fx_rates\): 'primary_key' column 'Rate_Date' is not in output_columns",
+    ):
+        parse_node(_upsert_entry(primary_key=["Rate_Date", "base"]), "c.yml")
+
+
+def test_primary_key_names_a_column_that_differs_only_in_case_from_another():
+    """With output columns base and Base, the key names exactly one of them."""
+    entry = _upsert_entry(primary_key=["rate_date", "Base"])
+    entry["output_columns"].append({"name": "Base", "type": "VARCHAR(3)", "nullable": False})
+    node = parse_node(entry, "c.yml")
+    assert node.primary_key == ("rate_date", "Base")

From a92fefa39f2ee718cef430b6b76bd43a5009ad24 Mon Sep 17 00:00:00 2001
From: Simone Carolini <simonecarolini.sc@gmail.com>
Date: Fri, 9 Oct 2026 13:25:31 +0200
Subject: [PATCH 3/5] feat(contract): emit mode_type and primary_key on the
 wire entry

mode_type appears only when it is not replace and primary_key only for
upsert, in authored order, so every existing entry and its config_hash and
content_hash stay byte-identical (pinned against a 0.8.0 known vector).

Signed-off-by: Simone Carolini <simonecarolini.sc@gmail.com>
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_0137E9XmyUXhwPTRrtNfivDw
---
 CHANGELOG.md                              |  4 +
 continuo_python_runtime/contract/merge.py | 13 +++-
 docs/boundary-contract.md                 | 13 ++++
 tests/contract/test_merge.py              | 91 ++++++++++++++++++++++-
 4 files changed, 117 insertions(+), 4 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 2c832ee..e97572e 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -14,6 +14,10 @@ follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
   `python-csv` nodes allow only `replace`. `continuo-runtime validate`, `merge`
   and `run` enforce these rules; the harness's write path for the new modes
   follows in a later entry.
+- The merged wire entry carries `mode_type` when it is not `replace` and
+  `primary_key` (authored order) for `upsert`; both are part of
+  `config_hash`. Entries of nodes that do not set them, and their hashes, are
+  unchanged.
 
 ### Changed
 
diff --git a/continuo_python_runtime/contract/merge.py b/continuo_python_runtime/contract/merge.py
index aa66fe2..7e9ec66 100644
--- a/continuo_python_runtime/contract/merge.py
+++ b/continuo_python_runtime/contract/merge.py
@@ -7,7 +7,7 @@
 
 from continuo_python_runtime.closure import resolve_closure
 from continuo_python_runtime.contract.loader import load_contract_dir
-from continuo_python_runtime.contract.model import CONTRACT_VERSION, Node
+from continuo_python_runtime.contract.model import CONTRACT_VERSION, DEFAULT_MODE_TYPE, Node
 from continuo_python_runtime.contract.paths import resolve_script_path
 from continuo_python_runtime.errors import ContractError
 from continuo_python_runtime.hashing import hash_parts
@@ -20,7 +20,10 @@ def node_entry(node: Node) -> dict:
     Returns the node as a dict with all fields, output_columns as list of
     dicts. The entry carries no hash fields: build_wire_contract adds all
     four (source_hash, shared_code_hash, config_hash, content_hash). The
-    nullable field is always present in output_columns.
+    nullable field is always present in output_columns. ``secret_ref``,
+    ``mode_type`` and ``primary_key`` appear only when they differ from their
+    default, so a node that does not use them has the entry (and hashes) it
+    had before they existed.
     """
     entry = {
         "schema": node.schema,
@@ -47,6 +50,12 @@ def node_entry(node: Node) -> dict:
     # only when set keeps every other node's config_hash unchanged.
     if node.secret_ref:
         entry["secret_ref"] = node.secret_ref
+    # Same reasoning: every replace node's entry, config_hash and content_hash
+    # stay byte-identical, so contract_version stays 1.
+    if node.mode_type != DEFAULT_MODE_TYPE:
+        entry["mode_type"] = node.mode_type
+    if node.mode_type == "upsert":
+        entry["primary_key"] = list(node.primary_key)
     return entry
 
 
diff --git a/docs/boundary-contract.md b/docs/boundary-contract.md
index b7f0898..83d9f11 100644
--- a/docs/boundary-contract.md
+++ b/docs/boundary-contract.md
@@ -61,6 +61,19 @@ s3://<bucket>/<service>/<release_id>/contract.yaml
   node's wire entry or `config_hash` changes), and it is part of
   `config_hash`, so `content_hash` changes when the referenced Secret name
   changes. `source_hash` does not.
+- **`mode_type`** and **`primary_key`** are additive and optional. `mode_type`
+  (`replace`, the default; `append`; `upsert`) says how the harness writes
+  the node's table on each run; `python-csv` allows only `replace`.
+  `primary_key` is required for `upsert` and rejected for every other mode: a
+  non-empty list of output column names, each spelled exactly as in
+  `output_columns` and declared `nullable: false`, unique case-insensitively. On the merged wire entry `mode_type` appears
+  only when it is not `replace`, and `primary_key` only for `upsert`, in the
+  authored order and spelling — so every existing node's wire entry,
+  `config_hash` and `content_hash` are unchanged and `contract_version` stays
+  1. Both are part of `config_hash` when present. A shared fixture of the
+  expected entries lives in this repo at
+  `tests/fixtures/wire_contract_write_modes.yml` and, byte-identical, in
+  continuo at `topology-controller/tests/fixtures/wire_contract_write_modes.yml`.
 - **Reads must be single-statement SELECTs with every table reference
   schema-qualified** (`analytics.table_a`, never `table_a`) — the resolver
   raises `UnqualifiedTableReference` and rejects the whole release otherwise
diff --git a/tests/contract/test_merge.py b/tests/contract/test_merge.py
index 34024fe..e161770 100644
--- a/tests/contract/test_merge.py
+++ b/tests/contract/test_merge.py
@@ -6,9 +6,14 @@
 
 import pytest
 
-from continuo_python_runtime.contract.merge import build_wire_contract, write_wire_contract
+from continuo_python_runtime.contract.loader import parse_node
+from continuo_python_runtime.contract.merge import (
+    build_wire_contract,
+    node_entry,
+    write_wire_contract,
+)
 from continuo_python_runtime.errors import ContractError
-from continuo_python_runtime.hashing import content_hash_fold
+from continuo_python_runtime.hashing import config_hash, content_hash_fold
 
 WIRE_ENTRY_KEYS = {
     "schema", "table", "owner", "schedule", "criticality", "script", "reads",
@@ -530,3 +535,85 @@ def test_python_api_script_is_held_to_the_closure_lint(tmp_path):
     (repo / "scripts" / "fx.py").write_text("import psycopg2\n\ndef run(ctx):\n    return None\n")
     with pytest.raises(ContractError, match="lint violations"):
         build_wire_contract(repo / "contracts", repo, "s")
+
+
+# --- write modes on the wire --------------------------------------------------
+
+
+def _modes_repo(tmp_path, **extra):
+    tmp_path.mkdir(parents=True, exist_ok=True)
+    (tmp_path / "contracts").mkdir()
+    (tmp_path / "scripts").mkdir()
+    (tmp_path / "scripts" / "fx.py").write_text(
+        "import pyarrow as pa\n\ndef run(ctx):\n    return pa.table({'rate': [1.0]})\n"
+    )
+    node = {
+        "schema": "fx", "table": "fx_rates", "owner": "m", "schedule": "hourly",
+        "criticality": "SECONDARY", "kind": "python-api", "script": "scripts/fx.py",
+        "output_columns": [
+            {"name": "rate_date", "type": "DATE", "nullable": False},
+            {"name": "base", "type": "VARCHAR(3)", "nullable": False},
+            {"name": "rate", "type": "DECIMAL(12,6)"},
+        ],
+    }
+    node.update(extra)
+    (tmp_path / "contracts" / "fx.yml").write_text(yaml.safe_dump({"nodes": [node]}))
+    return tmp_path
+
+
+def _only_entry(repo):
+    (entry,) = build_wire_contract(repo / "contracts", repo, "s")["nodes"]
+    return entry
+
+
+def test_replace_entry_is_identical_whether_or_not_mode_type_is_authored(tmp_path):
+    implicit = _only_entry(_modes_repo(tmp_path / "a"))
+    explicit = _only_entry(_modes_repo(tmp_path / "b", mode_type="replace"))
+    assert set(explicit) == WIRE_ENTRY_KEYS
+    assert explicit == implicit
+
+
+def test_append_entry_carries_mode_type_and_no_key(tmp_path):
+    entry = _only_entry(_modes_repo(tmp_path, mode_type="append"))
+    assert set(entry) == WIRE_ENTRY_KEYS | {"mode_type"}
+    assert entry["mode_type"] == "append"
+
+
+def test_upsert_entry_carries_its_key_in_authored_order(tmp_path):
+    entry = _only_entry(
+        _modes_repo(tmp_path, mode_type="upsert", primary_key=["base", "rate_date"])
+    )
+    assert set(entry) == WIRE_ENTRY_KEYS | {"mode_type", "primary_key"}
+    assert entry["mode_type"] == "upsert"
+    assert entry["primary_key"] == ["base", "rate_date"]
+
+
+def test_mode_type_and_primary_key_are_part_of_the_config_hash(tmp_path):
+    append = _only_entry(_modes_repo(tmp_path / "a", mode_type="append"))
+    upsert = _only_entry(
+        _modes_repo(tmp_path / "b", mode_type="upsert", primary_key=["rate_date", "base"])
+    )
+    reordered = _only_entry(
+        _modes_repo(tmp_path / "c", mode_type="upsert", primary_key=["base", "rate_date"])
+    )
+    assert append["source_hash"] == upsert["source_hash"] == reordered["source_hash"]
+    assert len({append["config_hash"], upsert["config_hash"], reordered["config_hash"]}) == 3
+
+
+# Computed with continuo-python-runtime 0.8.0 (before mode_type existed) for the
+# shared fixture's replace node. A replace entry must hash exactly like this.
+_PRE_WRITE_MODES_CONFIG_HASH = "651db2a6fdafda3f4dcdd5358132b0d7cd0177836be90380cd79388ac31d8cdf"
+_REPLACE_NODE = {
+    "schema": "fixture", "table": "replace_node", "owner": "data-team",
+    "schedule": "hourly", "criticality": "SECONDARY", "kind": "python-node",
+    "script": "scripts/replace_node.py", "reads": {"src": "SELECT id FROM fixture.src"},
+    "output_columns": [{"name": "id", "type": "BIGINT", "nullable": False}],
+    "description": "", "extra_columns": "raise", "config": {},
+}
+
+
+@pytest.mark.parametrize("extra", [{}, {"mode_type": "replace"}])
+def test_a_replace_entry_hashes_exactly_as_it_did_before_write_modes(extra):
+    entry = node_entry(parse_node({**_REPLACE_NODE, **extra}, "f.yml"))
+    assert "mode_type" not in entry and "primary_key" not in entry
+    assert config_hash(entry) == _PRE_WRITE_MODES_CONFIG_HASH

From 54116be5668cc8bba24f366db3b773b94be1aae6 Mon Sep 17 00:00:00 2001
From: Simone Carolini <simonecarolini.sc@gmail.com>
Date: Fri, 9 Oct 2026 13:27:45 +0200
Subject: [PATCH 4/5] test(contract): shared wire-contract fixture for write
 modes

Canonical wire entries for a replace, an append and an upsert node, without
hash fields. The real loader and node_entry reproduce each entry exactly; the
replace entry carries neither mode_type nor primary_key. continuo keeps a
byte-identical twin under topology-controller/tests/fixtures.

Signed-off-by: Simone Carolini <simonecarolini.sc@gmail.com>
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_0137E9XmyUXhwPTRrtNfivDw
---
 .../test_wire_contract_write_modes.py         | 53 +++++++++++++++++++
 tests/fixtures/wire_contract_write_modes.yml  | 53 +++++++++++++++++++
 2 files changed, 106 insertions(+)
 create mode 100644 tests/contract/test_wire_contract_write_modes.py
 create mode 100644 tests/fixtures/wire_contract_write_modes.yml

diff --git a/tests/contract/test_wire_contract_write_modes.py b/tests/contract/test_wire_contract_write_modes.py
new file mode 100644
index 0000000..f8d6ef8
--- /dev/null
+++ b/tests/contract/test_wire_contract_write_modes.py
@@ -0,0 +1,53 @@
+"""The shared write-modes wire fixture, reproduced by the real loader and node_entry().
+
+The fixture has a byte-identical twin in continuo
+(topology-controller/tests/fixtures/wire_contract_write_modes.yml); its header
+names both copies. Each fixture entry doubles as the author's contract node:
+parsing it and emitting it again must give back exactly the same entry.
+"""
+from pathlib import Path
+
+import pytest
+import yaml
+
+from continuo_python_runtime.contract.loader import parse_node
+from continuo_python_runtime.contract.merge import node_entry
+
+FIXTURE = Path(__file__).resolve().parents[1] / "fixtures" / "wire_contract_write_modes.yml"
+HASH_FIELDS = {"source_hash", "shared_code_hash", "config_hash", "content_hash"}
+
+
+def _fixture_nodes() -> list[dict]:
+    return yaml.safe_load(FIXTURE.read_text())["nodes"]
+
+
+@pytest.mark.parametrize("expected", _fixture_nodes(), ids=lambda entry: entry["table"])
+def test_node_entry_reproduces_the_fixture_entry(expected):
+    node = parse_node(dict(expected), FIXTURE.name)
+    assert node_entry(node) == expected
+
+
+def test_fixture_entries_carry_no_hash_fields():
+    for entry in _fixture_nodes():
+        assert not HASH_FIELDS & set(entry), entry["table"]
+
+
+def test_replace_entry_has_no_mode_keys_even_when_replace_is_authored():
+    (replace,) = [e for e in _fixture_nodes() if e["table"] == "replace_node"]
+    assert "mode_type" not in replace and "primary_key" not in replace
+    node = parse_node({**replace, "mode_type": "replace"}, FIXTURE.name)
+    assert node_entry(node) == replace
+
+
+def test_fixture_covers_every_mode_and_a_key():
+    entries = _fixture_nodes()
+    assert {e.get("mode_type", "replace") for e in entries} == {"append", "upsert", "replace"}
+    assert [e["primary_key"] for e in entries if "primary_key" in e] == [["rate_date", "base"]]
+
+
+def test_fixture_names_both_twins():
+    text = FIXTURE.read_text()
+    # Both twin paths are named in the header. Check each via a form that is not
+    # a substring of the other, so neither assertion is vacuous.
+    assert "topology-controller/tests/fixtures/wire_contract_write_modes.yml" in text
+    assert "continuo-python-runtime: tests/fixtures/wire_contract_write_modes.yml" in text
diff --git a/tests/fixtures/wire_contract_write_modes.yml b/tests/fixtures/wire_contract_write_modes.yml
new file mode 100644
index 0000000..2f3d054
--- /dev/null
+++ b/tests/fixtures/wire_contract_write_modes.yml
@@ -0,0 +1,53 @@
+# Shared wire-contract fixture for python node write modes.
+# Canonical wire entries as the runtime's node_entry() emits them, WITHOUT the
+# four hash fields (source_hash, shared_code_hash, config_hash, content_hash).
+# Twin copies, kept byte-identical:
+#   continuo:                topology-controller/tests/fixtures/wire_contract_write_modes.yml
+#   continuo-python-runtime: tests/fixtures/wire_contract_write_modes.yml
+nodes:
+  - schema: fixture
+    table: replace_node
+    owner: data-team
+    schedule: hourly
+    criticality: SECONDARY
+    kind: python-node
+    script: scripts/replace_node.py
+    reads:
+      src: SELECT id FROM fixture.src
+    output_columns:
+      - {name: id, type: "BIGINT", nullable: false}
+    description: ""
+    extra_columns: raise
+    config: {}
+  - schema: fixture
+    table: append_node
+    owner: data-team
+    schedule: hourly
+    criticality: SECONDARY
+    kind: python-api
+    script: scripts/append_node.py
+    reads: {}
+    output_columns:
+      - {name: collected_at, type: "TIMESTAMP", nullable: false}
+      - {name: rate, type: "DECIMAL(12,6)", nullable: true}
+    description: ""
+    extra_columns: raise
+    config: {}
+    mode_type: append
+  - schema: fixture
+    table: upsert_node
+    owner: data-team
+    schedule: hourly
+    criticality: SECONDARY
+    kind: python-api
+    script: scripts/upsert_node.py
+    reads: {}
+    output_columns:
+      - {name: rate_date, type: "DATE", nullable: false}
+      - {name: base, type: "VARCHAR(3)", nullable: false}
+      - {name: rate, type: "DECIMAL(12,6)", nullable: true}
+    description: ""
+    extra_columns: raise
+    config: {}
+    mode_type: upsert
+    primary_key: [rate_date, base]

From c587b883b7d0e4d9d8e0d1c25877af9c91c5f4be Mon Sep 17 00:00:00 2001
From: Simone Carolini <simonecarolini.sc@gmail.com>
Date: Fri, 9 Oct 2026 17:11:52 +0200
Subject: [PATCH 5/5] fix(harness): reject append/upsert until the write-modes
 dispatch lands

run_node() still writes every node with load() (replace). An accepted append
or upsert contract would therefore silently delete existing rows. Reject any
non-replace mode_type before the script runs or anything is written; the
per-mode dispatch in a later PR removes this guard.

Signed-off-by: Simone Carolini <simonecarolini.sc@gmail.com>
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_0137E9XmyUXhwPTRrtNfivDw
---
 continuo_python_runtime/harness.py | 12 ++++++++-
 tests/test_harness.py              | 39 ++++++++++++++++++++++++++++++
 2 files changed, 50 insertions(+), 1 deletion(-)

diff --git a/continuo_python_runtime/harness.py b/continuo_python_runtime/harness.py
index c86474c..0d797a3 100644
--- a/continuo_python_runtime/harness.py
+++ b/continuo_python_runtime/harness.py
@@ -29,7 +29,7 @@
 from continuo_python_runtime.conform import conform, to_arrow
 from continuo_python_runtime.context import RunContext
 from continuo_python_runtime.contract.loader import load_contract_dir
-from continuo_python_runtime.contract.model import Node
+from continuo_python_runtime.contract.model import DEFAULT_MODE_TYPE, Node
 from continuo_python_runtime.contract.paths import resolve_script_path
 from continuo_python_runtime.csv_loader import produce_csv
 from continuo_python_runtime.csv_source import CsvSourceReader
@@ -265,6 +265,16 @@ def run_node(
         nodes = load_contract_dir(contract_dir, check_reads=False)
         node = select_node(nodes, node_id)
 
+        # Interim: the harness writes every node with load() (replace) until the
+        # per-mode write dispatch lands. Until then, reject append/upsert before
+        # the script runs or anything is written, so an accepted contract cannot
+        # silently replace a table it meant to append to or upsert into.
+        if node.mode_type != DEFAULT_MODE_TYPE:
+            raise LoadError(
+                f"write mode {node.mode_type!r} is not supported by the harness yet "
+                f"(node {target_schema}.{table_name}); only {DEFAULT_MODE_TYPE!r} is available"
+            )
+
         if adapter is not None:
             active_adapter = adapter
         else:
diff --git a/tests/test_harness.py b/tests/test_harness.py
index 0c4a71a..e952115 100644
--- a/tests/test_harness.py
+++ b/tests/test_harness.py
@@ -715,3 +715,42 @@ def test_python_api_missing_env_key_is_a_script_error_naming_the_key(tmp_path, c
     assert run_node(_api_env(repo), adapter=FakeWarehouseAdapter()) == 1
     out = capsys.readouterr().out
     assert "ScriptError" in out and "FX_API_KEY" in out
+
+
+# --- non-replace write modes are rejected until the per-mode dispatch lands ---
+
+
+def _repo_with_mode(tmp_path, marker, **mode_fields):
+    repo = _repo_with_config(tmp_path, {}, marker)
+    doc = yaml.safe_load((repo / "contracts" / "t.yml").read_text())
+    del doc["nodes"][0]["config"]
+    doc["nodes"][0].update(mode_fields)
+    (repo / "contracts" / "t.yml").write_text(yaml.safe_dump(doc))
+    return repo
+
+
+@pytest.mark.parametrize(
+    "mode_fields",
+    [
+        {"mode_type": "append"},
+        {"mode_type": "upsert", "primary_key": ["id"]},
+    ],
+    ids=["append", "upsert"],
+)
+def test_run_node_rejects_non_replace_modes_before_writing_anything(
+    tmp_path, capsys, mode_fields
+):
+    """load() replaces the table, so an accepted append/upsert contract must not reach it."""
+    marker = tmp_path / "script-ran"
+    repo = _repo_with_mode(tmp_path, marker, **mode_fields)
+    ad = FakeWarehouseAdapter({"select id from analytics.a": pa.table({"id": [1]})})
+    assert run_node(_env(repo), adapter=ad) == 1
+    out = capsys.readouterr().out
+    assert out.count("===CONTINUO_VALIDATION_RESULT_BEGIN===") == 1
+    body = _sentinel_body(out)
+    assert body["status"] == "error"
+    assert body["message"].startswith("LoadError:")
+    assert "not supported by the harness yet" in body["message"]
+    assert mode_fields["mode_type"] in body["message"]
+    assert not marker.exists(), "script ran despite an unsupported write mode"
+    assert ad.ensured is None and ad.loaded is None