feat(contract): timeout_minutes, written outside config_hash

CHANGELOG.md

@@ -5,6 +5,15 @@ follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
 
 ## [Unreleased]
 
+### Added
+
+- Contracts may set `timeout_minutes` on any node: an integer from 1 to 10,080
+  that is the node's task timeout. Other values are refused as a contract
+  error naming the node. The merged wire entry carries the key only when set,
+  and it is the one wire key outside `config_hash`: changing it alone leaves
+  `config_hash` and `content_hash` unchanged. A node that does not set it has
+  the wire entry it always had.
+
 ### Changed
 
 - The "mode_type not allowed for kind" contract error now quotes the offending

README.md

@@ -176,6 +176,15 @@ produced, never how it is checked or written.
   script reads with `os.environ[...]`. The contract never holds a secret value
   or env var name; the Secret itself is created in the cluster by the operator.
 
+### Task timeout
+
+Any node may set `timeout_minutes:` — an integer from 1 to 10,080 (one week) —
+to declare how long its task may run before continuo gives up on it. A value
+outside that range, or one that is not an integer, fails
+`continuo-runtime validate`/`merge`/`lint`. It is the one key that is not part
+of `config_hash`, so changing only the timeout never changes the node's
+`content_hash`.
+
 ## Write modes
 
 `mode_type:` chooses how the harness writes a node's table on each run:

continuo_python_runtime/contract/loader.py

@@ -44,8 +44,13 @@
     "secret_ref",
     "mode_type",
     "primary_key",
+    "timeout_minutes",
 }
 
+# Bounds of a node's declared task timeout, in minutes (one minute to one week).
+MIN_TIMEOUT_MINUTES = 1
+MAX_TIMEOUT_MINUTES = 10_080
+
 _REQUIRED_STRING_FIELDS = ("schema", "table", "owner", "schedule")
 
 _ALLOWED_OUTPUT_COLUMN_KEYS = {"name", "type", "nullable"}
@@ -295,6 +300,16 @@ def parse_node(
             raise ContractError(f"{label}: 'secret_ref' is not allowed for kind {kind}")
         secret_ref = validate_secret_ref(raw["secret_ref"], label)
 
+    timeout_minutes = raw.get("timeout_minutes")
+    if "timeout_minutes" in raw and (
+        type(timeout_minutes) is not int
+        or not MIN_TIMEOUT_MINUTES <= timeout_minutes <= MAX_TIMEOUT_MINUTES
+    ):
+        raise ContractError(
+            f"{label}: 'timeout_minutes' must be an integer between "
+            f"{MIN_TIMEOUT_MINUTES} and {MAX_TIMEOUT_MINUTES}, got {timeout_minutes!r}"
+        )
+
     mode_type = raw.get("mode_type", DEFAULT_MODE_TYPE)
     if not isinstance(mode_type, str) or mode_type not in MODE_TYPES:
         raise ContractError(
@@ -375,6 +390,7 @@ def parse_node(
         secret_ref=secret_ref,
         mode_type=mode_type,
         primary_key=primary_key,
+        timeout_minutes=timeout_minutes,
     )
 
 

continuo_python_runtime/contract/merge.py

@@ -170,6 +170,10 @@ def build_wire_contract(
 
         primary_bytes, member_bytes = _HASH_INPUTS[node.kind](node, repo_root)
         entry.update(hash_parts(entry, primary_bytes, member_bytes))
+        # The one wire key outside config_hash: it is written after the hashes
+        # are computed, so changing it alone never changes content_hash.
+        if node.timeout_minutes is not None:
+            entry["timeout_minutes"] = node.timeout_minutes
 
         wire_nodes.append(entry)
 

continuo_python_runtime/contract/model.py

@@ -45,6 +45,7 @@ class Node:
     secret_ref: str = ""
     mode_type: str = DEFAULT_MODE_TYPE
     primary_key: tuple[str, ...] = ()
+    timeout_minutes: int | None = None
 
     @property
     def relation(self) -> str:

continuo_python_runtime/hashing.py

@@ -21,20 +21,25 @@
 
 HASH_PART_FIELDS = ("source_hash", "shared_code_hash", "config_hash")
 CONTENT_HASH_FIELD = "content_hash"
+# A wire key that is not part of the config_hash basis: it tunes how the node
+# runs (the task timeout) and says nothing about what the node computes, so
+# changing it alone is not a code change.
+NON_HASHED_FIELDS = ("timeout_minutes",)
 
 
 def canonical_entry(entry: dict) -> dict:
     """Deep-copy *entry*, drop the four hash fields, whitespace-normalize reads.
 
     Drops ``content_hash`` and all three of ``HASH_PART_FIELDS`` from the
     basis, and whitespace-normalizes every ``str`` value in ``reads`` via
-    ``" ".join(sql.split())``. Everything else is preserved verbatim,
-    including ``config``.
+    ``" ".join(sql.split())``. ``timeout_minutes`` is the one other wire key
+    left out of the basis (``NON_HASHED_FIELDS``). Everything else is
+    preserved verbatim, including ``config``.
     """
     canonical = deepcopy(entry)
 
     canonical.pop(CONTENT_HASH_FIELD, None)
-    for field in HASH_PART_FIELDS:
+    for field in (*HASH_PART_FIELDS, *NON_HASHED_FIELDS):
         canonical.pop(field, None)
 
     reads = canonical.get("reads")

docs/boundary-contract.md

@@ -61,6 +61,16 @@ 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.
+- **`timeout_minutes`** is additive and optional, valid on every `kind`: an
+  integer from 1 to 10,080 (one week) that is the node's task timeout. Any
+  other value (zero, above the bound, a string, a float, a boolean) fails
+  `ContractError` naming the node and the key. On the merged wire entry it
+  appears only when the node declares it, and it is the one wire key **outside
+  `config_hash`**: it is written after the hash parts are computed and is
+  excluded from the `config_hash` basis, so changing it alone changes none of
+  the four hashes and is not a code change. A node that does not declare it
+  has the wire entry, `config_hash` and `content_hash` it always had, and
+  `contract_version` stays 1.
 - **`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`.
@@ -222,7 +232,8 @@ content_hash     = "sha256:" + sha256(source_hash + "|" + shared_code_hash
   collapse to a single space, then stripped); `content_hash` **and all
   three part fields** (`source_hash`, `shared_code_hash`, `config_hash`)
   are excluded from the basis — `config_hash` cannot include itself, and
-  none of the four hash fields participates in another's basis. Every
+  none of the four hash fields participates in another's basis. So is
+  `timeout_minutes`, the one non-hash wire key left out of the basis. Every
   other field, including `config`, participates verbatim.
 
 Invariant: formatting-only edits (spaces in SQL, yaml indentation/key

tests/contract/test_loader.py

@@ -878,3 +878,22 @@ def test_primary_key_names_a_column_that_differs_only_in_case_from_another():
     entry["output_columns"].append({"name": "Base", "type": "VARCHAR(3)", "nullable": False})
     node = parse_node(entry, "c.yml")
     assert node.primary_key == ("rate_date", "Base")
+
+
+# --- timeout_minutes ------------------------------------------------------------
+
+
+def test_timeout_minutes_is_read():
+    assert parse_node({**VALID, "timeout_minutes": 120}, "f.yml").timeout_minutes == 120
+    assert parse_node({**VALID, "timeout_minutes": 1}, "f.yml").timeout_minutes == 1
+    assert parse_node({**VALID, "timeout_minutes": 10080}, "f.yml").timeout_minutes == 10080
+
+
+def test_timeout_minutes_defaults_to_none():
+    assert parse_node(VALID, "f.yml").timeout_minutes is None
+
+
+@pytest.mark.parametrize("bad", [0, 10081, "2h", 1.5, True])
+def test_a_bad_timeout_minutes_is_refused(bad):
+    with pytest.raises(ContractError, match=r"f\.yml \(analytics\.t\): 'timeout_minutes'"):
+        parse_node({**VALID, "timeout_minutes": bad}, "f.yml")

tests/contract/test_merge.py

@@ -617,3 +617,25 @@ 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
+
+
+# --- timeout_minutes ----------------------------------------------------------
+
+
+def test_timeout_minutes_is_outside_the_config_hash(tmp_path):
+    without = _only_entry(_modes_repo(tmp_path / "a"))
+    short = _only_entry(_modes_repo(tmp_path / "b", timeout_minutes=120))
+    long = _only_entry(_modes_repo(tmp_path / "c", timeout_minutes=480))
+    assert short["config_hash"] == long["config_hash"] == without["config_hash"]
+    assert short["content_hash"] == long["content_hash"] == without["content_hash"]
+    # The wire entry hashes exactly as the same entry would without the key.
+    assert config_hash(short) == short["config_hash"]
+
+
+def test_the_wire_entry_carries_timeout_minutes_only_when_set(tmp_path):
+    without = _only_entry(_modes_repo(tmp_path / "a"))
+    declared = _only_entry(_modes_repo(tmp_path / "b", timeout_minutes=120))
+    assert "timeout_minutes" not in without
+    assert set(without) == WIRE_ENTRY_KEYS
+    assert declared["timeout_minutes"] == 120
+    assert set(declared) == WIRE_ENTRY_KEYS | {"timeout_minutes"}

tests/test_hashing.py

@@ -186,3 +186,10 @@ def test_hash_parts_returns_four_keys_and_consistent_fold():
     assert parts["content_hash"] == content_hash_fold(
         parts["source_hash"], parts["shared_code_hash"], parts["config_hash"]
     )
+
+
+def test_timeout_minutes_is_outside_the_config_hash_basis():
+    with_timeout = {**ENTRY, "timeout_minutes": 120}
+    assert "timeout_minutes" not in canonical_entry(with_timeout)
+    assert config_hash(with_timeout) == config_hash(ENTRY)
+    assert config_hash({**ENTRY, "timeout_minutes": 480}) == config_hash(ENTRY)