diff --git a/conformance/fixtures/README.md b/conformance/fixtures/README.md index 17f87a5..914f1d0 100644 --- a/conformance/fixtures/README.md +++ b/conformance/fixtures/README.md @@ -14,8 +14,16 @@ than a measurement. These three documents turn it back into a measurement: away from the real parser fails on the real parser's own reference documents. What they exercise that this harness's own traffic does not: a `layout='prefix'` input -port with a real `prefix_digest`, `organization_id`, `upstream_execution_id`, a `null` -`prefix_digest` on a file port, and a marker delivering two different ports. +port with a real `prefix_digest`, an integer `organization_id`, and a marker delivering +two different ports. + +What they do **not** cover: `upstream_execution_id` as `null`. All three objects in the +golden manifest carry an integer there (89990), and `null` is just as real: it is what an +object carries whenever no producing execution is known for it, as on a run an operator +started by hand. That form is built by `tests/test_harness_self.py`, for this harness's +copy of the contract, and by `tests/test_inputs.py`, for the node. (The harness's own job +descriptions now write the field on every object, and a `null` `prefix_digest` on every +file port, as the launcher does. Both used to be things only these documents contained.) Do not edit them. If the orchestrator's goldens change, re-copy them and read what changed — that is the signal, not a merge conflict to resolve. diff --git a/conformance/job.py b/conformance/job.py index cf8e9bd..861c5bc 100644 --- a/conformance/job.py +++ b/conformance/job.py @@ -32,6 +32,17 @@ #: that keeps a superseded runner out of the live attempt's directory. BUCKET = 'humansignal-orchestrator-conformance' +#: The kind every port carries unless a test names a real one. Invented on purpose: the +#: orchestrator never emits it, so a node that started to depend on a port's kind could +#: not pass here by coincidence. +DEFAULT_PAYLOAD_KIND = 'conformance_bytes' + +#: The execution every input object names as its producer unless a test says otherwise — +#: an integer, which is what a run started by an upstream step's completion carries. The +#: other real value is ``None``: the launcher copies the id out of the run's own trigger +#: and writes ``null`` when there is none there, as on a run an operator started by hand. +DEFAULT_UPSTREAM_EXECUTION_ID = 4241 + @dataclass class InputSpec: @@ -106,6 +117,8 @@ def __init__( input_extra: dict | None = None, max_object_bytes: int = 256 * 1024 * 1024, creds_ttl_s: float | None = None, + payload_kind: str = DEFAULT_PAYLOAD_KIND, + upstream_execution_id: int | None = DEFAULT_UPSTREAM_EXECUTION_ID, ): self.image = image self.workdir = workdir @@ -124,6 +137,12 @@ def __init__( # expire, but fifteen minutes is longer than any test here runs, so an unbounded # one models the ordinary case exactly and only the expiry tests shorten it. self.creds_ttl_s = creds_ttl_s + # One of each per JOB, not per object, because that is how the launcher builds + # them (``handlers/steps/external.py``): the port takes a single kind, and the one + # execution id the run's trigger carried — or ``None`` — is copied onto every + # object. A per-object knob would let a test describe a job production cannot send. + self.payload_kind = payload_kind + self.upstream_execution_id = upstream_execution_id self.idempotency_key = f'exec-{execution_id}-att-{attempt}' self.key_prefix = ( @@ -152,7 +171,20 @@ def __init__( # ------------------------------------------------------------------ documents def manifest(self) -> dict: - """``invocation.json`` — the job description, exactly as the launcher writes it.""" + """``invocation.json`` — the job description, with every field the launcher writes. + + **A field that is present and ``null`` is a shape of its own, and production sends + it.** The orchestrator serialises this document without dropping empty values + (``external/io.py`` dumps the whole model), so a real input object ALWAYS carries + ``upstream_execution_id`` — an integer, or ``null`` — a real file port ALWAYS + carries ``"prefix_digest": null``, and the document itself always carries + ``organization_id``. This method used to leave all three out, which made "the key + is there and its value is null" something no container in this suite had ever + been handed, although every real job with an input hands over at least one. + + ``organization_id`` is ``null`` here: a synthetic job belongs to no organization. + The integer form is in the golden manifest under ``conformance/fixtures/``. + """ ports: dict[str, dict] = {} for spec in self.inputs: pin = spec.pin() @@ -160,10 +192,11 @@ def manifest(self) -> dict: spec.port, { 'name': spec.port, - 'payload_kind': 'conformance_bytes', + 'payload_kind': self.payload_kind, 'layout': 'file', 'cardinality': 'many', 'objects': [], + 'prefix_digest': None, }, ) port['objects'].append( @@ -172,12 +205,14 @@ def manifest(self) -> dict: 'sha256': pin.sha256, 'size': pin.size, 'relpath': spec.relpath, + 'upstream_execution_id': self.upstream_execution_id, } ) document = { 'schema_version': 1, 'execution_id': self.execution_id, 'pipeline_id': 9, + 'organization_id': None, 'deployment_id': 3, 'revision_id': 5, 'image_digest': self.image, diff --git a/docs/AUTHORING.md b/docs/AUTHORING.md index 2bdc78d..bfa75a0 100644 --- a/docs/AUTHORING.md +++ b/docs/AUTHORING.md @@ -16,8 +16,11 @@ which is what keeps a mistake in step 3 from being discovered in step 5. ### 1. Decide what your node consumes and produces -**BEHAVIOUR.** A job with upstream artifacts receives them on one port named `input`; -a source-style job receives no input ports. Your node returns objects grouped into output +**BEHAVIOUR.** A job with upstream artifacts receives them on one port named `input`. A +job without them, whose node is wired directly after a Data Source node, receives the file +an operator uploaded there on the same port, as one object +([PROTOCOL.md](PROTOCOL.md#22-invocationjson-field-by-field)). A source-style job, with +neither, receives no input ports. Your node returns objects grouped into output ports you name yourself. Those port names become the artifact kinds downstream steps ask for ([PROTOCOL.md](PROTOCOL.md#41-where-your-output-goes)). @@ -496,11 +499,25 @@ apart either treats advice as law or treats law as advice. Both are expensive. can catch a changed input. * [ ] **RECOMMENDATION.** Handles a job with **no** input port at all — an empty `inputs` list is legitimate, not an error. +* [ ] **RECOMMENDATION.** Does not rely on the input port's `payload_kind`, or on an + object's `upstream_execution_id`. The kind is a label and nothing checks the bytes + against it: it is the first upstream artifact's kind, or, for a file an operator + uploaded to a Data Source, the last extension of the file's cleaned stored name. The + execution id is `null` whenever no execution id is known: a Data Source upload, or a + run of this step that was started by hand rather than by an upstream step completing + (`handlers/steps/external.py`). +* [ ] **RECOMMENDATION.** Takes an uploaded file under whatever name it arrives with, not + a name agreed out of band. The name is cleaned when the file is uploaded — every run + of characters outside `A-Za-z0-9._-` becomes one `_` — so a file whose name holds + anything else, a space or an accented letter included, reaches the container under a + different one (`résumé 2026.csv` arrives as `r_sum_2026.csv`), and an archive arrives + as one object, not unpacked. * [ ] **RECOMMENDATION.** Handles two inputs that share a basename. * [ ] **RECOMMENDATION.** Bounds its read of the job description, and refuses a `schema_version` it does not implement. -* [ ] **RECOMMENDATION.** Deletes each input when it is done with it. Nothing bounds input - bytes; the manifest bounds count indirectly. The container has no disk quota. +* [ ] **RECOMMENDATION.** Deletes each input when it is done with it. Nothing in the + contract or the agent bounds input bytes; the manifest bounds count indirectly. The + container has no disk quota. #### Work diff --git a/docs/CONFORMANCE.md b/docs/CONFORMANCE.md index 3dba31b..720feea 100644 --- a/docs/CONFORMANCE.md +++ b/docs/CONFORMANCE.md @@ -447,7 +447,10 @@ that ignores SIGTERM completely. Assert the marker and the timing. ## Level 3: the real thing **RECOMMENDATION.** Register the node, start an agent, run the pipeline. See -[OPERATIONS.md](OPERATIONS.md#registering-a-node). +[OPERATIONS.md](OPERATIONS.md#registering-a-node). A first run needs no upstream step: +upload a test file to a Data Source node, wire that node straight into yours and start the +run with **Run node** on your node +([OPERATIONS.md](OPERATIONS.md#giving-the-node-a-file-you-uploaded)). **What only level 3 proves.** That your container can read the credentials a **real** agent wrote — which is not a question about your image's uid (any uid can open them) but about diff --git a/docs/OPERATIONS.md b/docs/OPERATIONS.md index 82f860a..c00afd8 100644 --- a/docs/OPERATIONS.md +++ b/docs/OPERATIONS.md @@ -361,21 +361,27 @@ staging directory. ## Pointing a pipeline node at your deployment -**BEHAVIOUR.** In the pipeline editor: **Add node → External → External node**, pick the deployment you -registered, and wire an edge into it from the step whose output it should read. That is -the same thing as adding a script node with this configuration, which is what it stores -and what an API caller writes directly: +**BEHAVIOUR.** In the pipeline editor: **Add node → External → External node**, and pick +the deployment you registered. Wire an edge into it from whatever should feed it: the step +whose output it should read, or a **Data Source** node holding an uploaded file. Use the +`MANUAL` trigger on an edge from a Data Source (a newly drawn one has it by default) and +start the run with the **Run node** button on the external node itself — a Data Source has +no Run button of its own. Whatever the node delivers through its output ports comes back as +that run's artifacts. The Data Source route has a section of its own: +[Giving the node a file you uploaded](#giving-the-node-a-file-you-uploaded). + +**BEHAVIOUR.** Adding the node in the editor is the same thing as adding a script node with +this configuration, which is what the editor stores and what an API caller writes directly: ```json {"step_kind": "external", "external_deployment_id": 12, "params": {}} ``` -**BEHAVIOUR.** Whatever the node delivers through its output ports comes back as that -run's **artifacts**, and each one's kind is the **port name** it was delivered under. A -downstream step configured to read `output` finds exactly what was published under -`output` — which is why "the node ran but nothing downstream sees anything" is almost -always a port name that does not match, or an object the marker inventoried but claimed -under no port at all. +**BEHAVIOUR.** Each artifact the node delivers takes its kind from the **port name** it was +delivered under. A downstream step configured to read `output` finds exactly what was +published under `output` — which is why "the node ran but nothing downstream sees +anything" is almost always a port name that does not match, or an object the marker +inventoried but claimed under no port at all. **RULE for the "Required" column, BEHAVIOUR for the rest of the table below.** The configuration validator refuses a node whose `external_deployment_id` is missing or is not @@ -389,7 +395,7 @@ did set (`pipelines/config_schemas.py`). | `params` | object | no, default `{}` | **the only node configuration your code ever sees**, copied verbatim | **yes** | | `timeout_seconds` | integer > 0 or null | no | requested runtime budget. **Set this to give a node more than the 900 seconds its revision declares** — it wins over the revision and takes effect at once | **yes** | | `queue_timeout_seconds` | integer > 0 or null | no | how long the job stays claimable, default 3600 | no | -| `input_payload_kind` | string or null | no | require an upstream artifact of this kind; absent means every upstream artifact, including none | indirectly | +| `input_payload_kind` | string or null | no | shown in the editor as **Upstream artifact kind**. Require an input of this kind: an upstream artifact of that kind, or, when the input is a Data Source's uploaded file, a file of that kind — the last extension of its cleaned stored name, lower-cased (`zip` for `batch.zip`), or `file` when it has none. Kinds are compared exactly, and a file's kind is always lower case, so write `zip`, not `ZIP`. Absent means whatever is there: every upstream artifact, or the Data Source's file, including nothing at all | indirectly | **RULE.** New API writes reject unknown top-level external-step configuration keys. A typo such as `timeout_second` cannot be saved as a supported setting. Put the @@ -397,14 +403,78 @@ container's own options inside `params`, whose shape the platform does not valid Older stored nodes with stray keys can still run; the handler ignores those keys (`pipelines/config_schemas.py`, `handlers/steps/external.py`). -**BEHAVIOUR.** With `input_payload_kind` set and no matching upstream artifact, the step -fails after exhausting the engine's ordinary retry ladder rather than immediately. With it -unset and no upstream artifacts at all, the job runs with no input port, which is a -legitimate source-style step. +**BEHAVIOUR.** With `input_payload_kind` set and the node fed by an upstream **step**, a +run with no upstream artifact of that kind is the ordinary missing-artifact failure: the +step fails after exhausting the engine's ordinary retry ladder rather than immediately, +because the upstream step may still produce the artifact. + +**RULE.** With `input_payload_kind` set and the input a **Data Source's uploaded file** — +the node is wired directly after a Data Source and the run carries no upstream artifacts — +the run fails **immediately** when the file is of another kind, or when the Data Source +holds no uploaded file at all (`handlers/steps/external.py`). That failure is terminal: +nothing is retried and no job is queued, because no amount of waiting uploads a file. When +the run does carry upstream artifacts, the paragraph above applies instead. + +**BEHAVIOUR.** With it unset, the node gets whatever is there: every upstream artifact, or +the Data Source's uploaded file. When there is neither, the job runs with no input port, +which is a legitimate source-style step. **BEHAVIOUR.** A folder (prefix layout) artifact upstream is refused immediately and permanently, with a message naming the artifact. It is not retried. +### Giving the node a file you uploaded + +**BEHAVIOUR.** An external node wired directly after a **Data Source** node — a pipeline +node that holds files an operator uploads in the editor — receives that Data Source's +uploaded file as its one input, with no Import step and no other step in between, whenever +the run carries no upstream artifacts (`handlers/steps/external.py`). + +1. Add a **Data Source** node, open it and upload the file. The newest upload is the + **Active** one unless you chose another with **Make active**. +2. Draw an edge from the Data Source to the external node. It has to be on the `MANUAL` + trigger: an edge newly drawn from a Data Source has that by default, and an edge drawn + earlier keeps the trigger it was given, so check it. +3. Start the run with the **Run node** button on the **external node**. A Data Source has + no Run button of its own. + +**BEHAVIOUR.** Which file a run gets is settled at that click. It is the one that was +Active when Run was clicked (pinned at that moment), if it is still an uploaded file of that +Data Source; otherwise the Data Source's active file, which is the one explicitly made +Active, else the newest upload. While the upload pinned at the click still exists, a file +uploaded afterwards does not replace it: a **retry** keeps the file the run was started +with, and a newer upload reaches the node only through a fresh Run. When that upload has +been deleted, the Data Source's active file is used instead and a warning is logged, and if +the Data Source then holds no file the retry launches with zero inputs unless a kind is +required. A run started while the Data Source held no file pinned nothing, so a retry of it +reads whatever file is active by then. + +**BEHAVIOUR.** The container is handed one object on the `input` port, pinned by `sha256` +and `size` and downloaded through a presigned link like any other input. Its name is +the file's cleaned stored name — every run of characters outside `A-Za-z0-9._-` became one +`_` at upload — and the port's kind is that name's last extension, lower-cased, or `file` +when it has none ([PROTOCOL.md](PROTOCOL.md#22-invocationjson-field-by-field)). The run's +log has a line naming the Data Source, the file as it was uploaded and its kind, and its +start line reports `inputs=1`. + +**BEHAVIOUR, for every bullet below.** These are the limits of it. + +* **Uploaded files only.** An S3 address typed into the Data Source's settings is not + delivered to an external node. +* **One file per run, and never alongside upstream artifacts.** A run that carries upstream + artifacts gets those, handled and retried as for any other source, and the uploaded file + is not added. +* **No file is not an error, unless a kind is required.** With no uploaded file on the Data + Source and `input_payload_kind` unset, the job launches with `"inputs": []`. With the + setting present the run fails at once, as described above. +* **Uploads are capped at 500 MB by default.** That is a limit on the **declared** size of + an upload, and an installation can change it. +* **Deleting an upload does not stop a hand-over.** It removes the upload's row, not the + stored object, so a job already queued still receives the file. Cancelling the run is the + stop control. +* **External nodes only, and no change to the wire contract.** No other kind of step is + handed the file this way, and the job description is still version 1 with no field added + ([PROTOCOL.md](PROTOCOL.md#22-invocationjson-field-by-field)). + --- ## Revisions: what is immutable and what is actually read @@ -493,7 +563,9 @@ which you cannot change from a node. * **No aggregate upload quota.** One object is bounded at 1 GiB. Nothing caps the number of objects or the total bytes a container may upload before its credentials expire. * **No limit on the SIZE of inputs, and an indirect one on their number.** Nothing caps one - input object or their total bytes. The count, however, is capped after all: every input + input object or their total bytes — except at the one place an operator uploads an input + directly: a file uploaded to a Data Source is capped at 500 MB by default, a limit on the + declared size of the upload. The count, however, is capped after all: every input and its metadata are written into the job description, and the writer refuses to publish one over 8 MiB (`external/io.py`). Measured against the real models, that is roughly **25,000 inputs** with 95-character URIs, at about 330 bytes each. A step wired @@ -618,6 +690,10 @@ two rows where that distinction bites are marked inline. | Logs never appear at all | a logging library that defaults to WARNING and to stderr only, or a buffered stdout | Unchecked | configure the logger explicitly and set `PYTHONUNBUFFERED=1` | | A container's output is gone before you could read it | the agent removes a job's container when the job ends | BEHAVIOUR — exited containers are removed unless the agent enables `LSPO_AGENT_KEEP_CONTAINERS` | attach at the start instead: [Watching a run, live](#watching-a-run-live) | | Log lines arrive mangled, with stray `[31m` in them | you printed ANSI colour; the escape byte is stripped as a control character and the rest survives | BEHAVIOUR | print plain text | +| The node is wired straight after a **Data Source** node and starts with no inputs (`"inputs": []`), or with inputs that are not the uploaded file | no inputs: the Data Source holds no uploaded file, or the orchestrator predates this capability. Only an uploaded file is delivered, never an S3 address typed into the Data Source's settings; and a retry launches with none when the file the run was started with has been deleted and the Data Source holds no other. Other inputs: the run carried upstream artifacts, which are then the node's inputs, and the uploaded file is not added to them | BEHAVIOUR — a job with no inputs is legitimate, so nothing refuses it | upload the file on the Data Source, check that it is the **Active** one, then start a fresh run with **Run node** on the external node. The run's log names the file that was handed over, and its start line reports `inputs=1`. Nothing to fix in the node: the empty list is the platform's answer, not a fault in your reader ([above](#giving-the-node-a-file-you-uploaded)) | +| The external node is wired after a Data Source and never starts, or shows no **Run node** button | the edge from the Data Source is on the "upstream complete" trigger. That trigger fires when an execution **into** the edge's source node completes, and a Data Source with no edge into it never has one. A node shows its Run button only for an incoming edge on the manual trigger. An edge newly drawn from a Data Source has that trigger by default; an edge drawn earlier keeps the one it was given | BEHAVIOUR — the trigger is honoured as set, and no job is created | switch that edge to the manual (`MANUAL`) trigger in the pipeline editor, then start the run with **Run node** on the external node — a Data Source has no Run button of its own. Nothing to fix in the node or the agent: neither was asked to do anything, so neither has a log line about it | +| The run retries for a long time, then fails naming the upstream node and a kind | **Upstream artifact kind** (`input_payload_kind`) is set on the node, and the upstream **step** produced no artifact of that kind | BEHAVIOUR — the run fails only after the engine's ordinary retry ladder, not at once ([above](#pointing-a-pipeline-node-at-your-deployment)) | compare the setting with what the upstream step really publishes. When that step is itself an external node, the kind is the **port name** its output was delivered under | +| The run fails at once, with no job queued, naming a Data Source and the kind of its file, or saying the Data Source has no uploaded file | **Upstream artifact kind** (`input_payload_kind`) is set on a node whose input is a Data Source's uploaded file, and that file is of another kind — its kind is the last extension of its cleaned stored name, lower-cased, or `file` when it has none — or the Data Source holds no uploaded file at all | RULE — checked before the job is created, and not retried (`handlers/steps/external.py`) | make the setting match the file's kind exactly (`zip` for `batch.zip`, never `ZIP`), or clear it to accept any file, or upload a file of the required kind and make it Active. Then start a fresh run: a retry keeps the file the run was started with for as long as that upload exists | | The node runs but downstream steps see nothing | the objects were inventoried but claimed under no output port | BEHAVIOUR — an unclaimed object is verified and kept, but not offered downstream | claim them under a port; `produced_ports` is what becomes downstream artifacts | | Saving external-node configuration rejects a key | unknown top-level configuration key | RULE — external configuration is closed at the API write boundary | check the table above; put container-specific options inside `params` | | The disk on the agent's machine fills up | inputs downloaded and never deleted; no storage limit exists | Unchecked | delete each input when done; see the residual limits above | diff --git a/docs/PROTOCOL.md b/docs/PROTOCOL.md index 26b7ae0..a361a51 100644 --- a/docs/PROTOCOL.md +++ b/docs/PROTOCOL.md @@ -326,7 +326,7 @@ An input port (`external/contract.py`): | Field | Type | Required | Notes | |---|---|---|---| | `name` | string | **yes** | the port name your code refers to. Today this is always `input` | -| `payload_kind` | string | **yes** | logical content type. Taken from the **first** upstream artifact's own kind, or the literal string `file` when that artifact carries no kind at all. It is not the node's `input_payload_kind` setting, which only selects which artifact gets pinned | +| `payload_kind` | string | **yes** | logical content type: a label, which nothing checks the bytes against. For the artifacts of an upstream step it is taken from the **first** artifact's own kind, or is the literal string `file` when that artifact carries no kind at all. For a file uploaded to a Data Source (below) it is the last extension of the file's cleaned stored name, lower-cased and without the dot (`batch.zip` gives `zip`, `rows.CSV` gives `csv`, `x.tar.gz` gives `gz`), or `file` when that name has no extension, has its only dot as the first character (`.env`; `.data.csv` still gives `csv`), or ends in a dot. It is not the node's `input_payload_kind` setting, which only decides what is required: which upstream artifact gets pinned, or the kind an uploaded file must have | | `layout` | `"file"` or `"prefix"` | **yes** | today always `"file"` | | `cardinality` | `"one"`, `"at_least_one"` or `"many"` | **yes** | today always `"many"`, and it is enforced by the parser | | `objects` | array | no, defaults to `[]` | | @@ -340,13 +340,59 @@ An input object (`external/contract.py`): | `sha256` | 64 lowercase hex characters | **yes** | | | `size` | integer >= 0 | **yes** | | | `relpath` | canonical relative path, or `null` | no for a file port, **yes** for every object of a prefix port | | -| `upstream_execution_id` | integer >= 1, or `null` | no | which execution produced this object | +| `upstream_execution_id` | integer >= 1, or `null` | no | which execution produced this object. It is `null` whenever no execution id is known: a Data Source upload, or a run of this step that was started by hand rather than by an upstream step completing. Both are ordinary runs, so `null` is a normal value and not a fault | -**BEHAVIOUR.** A step with no upstream artifacts gets `"inputs": []`, an empty list with +**BEHAVIOUR.** A step with nothing to read gets `"inputs": []`, an empty list with **no ports at all**, rather than one port with no objects -(`handlers/steps/external.py`). Your code must handle a job with no input port -present, not merely a port that is empty. A step with no inputs is a legitimate -source-style step and is not an error. +(`handlers/steps/external.py`). That is a step whose run carries no upstream artifacts +and which is handed no uploaded file either (next paragraph). Your code must handle a job +with no input port present, not merely a port that is empty. A step with no inputs is a +legitimate source-style step and is not an error. + +**BEHAVIOUR.** The objects on `input` come from one of two places, never both in one run, +and nothing in their shape says which. Normally they are the artifacts of the step your +node is wired after. When your node is wired directly after a **Data Source** node — a +pipeline node that holds files an operator uploads in the editor — and the run carries no +upstream artifacts, the input is that Data Source's uploaded file instead: **one** object +on the same `input` port, pinned by `sha256` and `size` like any other input and +downloaded through the same kind of presigned link in the credentials envelope +(`handlers/steps/external.py`). Your reading code does not change. What is particular to +that object: + +* **Which file it is.** The one that was Active on the Data Source when Run was clicked + (it is pinned at that moment), if it is still an uploaded file of that Data Source; + otherwise the Data Source's active file, which is the one explicitly made Active, else + the newest upload (`pipelines/source_assets.py`). While the upload pinned at the click + still exists, a file uploaded afterwards does not replace it for that run; when nothing + was pinned (the Data Source held no file at the click) or the pinned upload was deleted, + the run reads the file that is active when it is launched or retried. +* **Its name.** `relpath` is the file's **cleaned** stored name, which is not necessarily + the name the operator sees: every run of characters outside `A-Za-z0-9._-` became one + `_` when the file was uploaded, so `sample 100.zip` arrives as `sample_100.zip`. +* **Its kind.** The port's `payload_kind` is read from that same cleaned name, by the rule + in the port table above — so the name and the kind your container sees always agree. +* **Its provenance.** `upstream_execution_id` is `null`, as it is whenever no execution id + is known: a Data Source upload, or a run of this step that was started by hand rather + than by an upstream step completing. +* **Nothing unpacks it.** The file arrives as it was uploaded. An archive is one object. + +**BEHAVIOUR, and these are the limits of it.** Uploaded files only: an S3 address typed +into the Data Source's settings is not delivered. External nodes only: no other kind of +step is handed the file this way. One file per run, and never alongside upstream artifacts +— when the run does carry upstream artifacts they are the inputs, handled exactly as for +any other source, and the uploaded file is not added. A Data Source holding no uploaded +file hands over nothing, and the job is then the zero-input job described above — unless +the node's `input_payload_kind` setting requires a kind, in which case the run fails +before any job exists and your container never starts +([OPERATIONS.md](OPERATIONS.md#giving-the-node-a-file-you-uploaded)). The wire contract is +unchanged: this is still version 1, and no field was added to it. + +**RECOMMENDATION.** Take whatever arrives on the port. Do not look an uploaded file up by a +name agreed out of band, because the name your container sees is the cleaned one; do not +treat `payload_kind` as proof of what the bytes are, because for an upload it is only part +of a file name; and do not require `upstream_execution_id` to be an integer. Nothing +checks any of the three, so a node that leans on one of them fails on a perfectly +ordinary upload, with no refusal to explain why. **BEHAVIOUR.** `credentials_file` in the job description is an echo, not a discovery mechanism. It cannot be one: you needed the credentials to fetch this document. The @@ -379,6 +425,14 @@ orchestrator sets an input's relpath to the last segment of its URI the uniqueness check exists only for folder ports (`external/contract.py`). Two upstream steps that both produce `rows.csv` therefore collide. +**BEHAVIOUR.** For a file uploaded to a Data Source (section 2.2) that last segment is the +**cleaned** stored name, which is not necessarily the name the operator chose. Every run +of characters outside `A-Za-z0-9._-` was replaced by one `_` when the file was uploaded +(`storage_connectors/services.py`), so `My Data (final).zip` arrives as +`My_Data_final_.zip` and `résumé.zip` as `r_sum_.zip`. The run's log shows the name as it +was uploaded; your container sees the cleaned one, and the port's kind is read from the +cleaned one too. + **RULE.** Your output relpaths must be unique within one marker (`external/contract.py`). @@ -390,10 +444,14 @@ end of the run, after all the work. **RECOMMENDATION.** Stream inputs to disk or process them incrementally. See section 3.5 for why holding one in memory is a real risk rather than a style preference. -**BEHAVIOUR.** There is no ceiling on the **size** of your inputs: not on one object, not -on their total. The 1 GiB ceiling in section 3.5 applies to what you **upload**, and the -container is given no disk quota at all — the agent sets a memory limit, a CPU limit and a -process limit when it starts your container, and no storage limit +**BEHAVIOUR.** Nothing in the contract or the agent puts a ceiling on the **size** of your +inputs: not on one object, not on their total. One source of input has a limit of its own, +applied long before your job exists: a file uploaded to a Data Source is capped at 500 MB +by default (`pipelines/asset_views.py`). That is a limit on the size **declared** when the +upload starts, and an installation can change it, so the pinned `size` of the object is the +number to read, not the cap. The 1 GiB ceiling in section 3.5 applies to what you +**upload**, and the container is given no disk quota at all — the agent sets a memory +limit, a CPU limit and a process limit when it starts your container, and no storage limit (`agent/executors/docker_exec.py`). A node that downloads every input to `/tmp` and keeps it there can therefore fill the disk of the machine the agent runs on, which is somebody else's laptop or server. Delete each input when you are done with it, or stream it diff --git a/docs/README.md b/docs/README.md index 4158a53..c5ca757 100644 --- a/docs/README.md +++ b/docs/README.md @@ -120,6 +120,8 @@ own label. | How to register my node and start the agent | [OPERATIONS.md](OPERATIONS.md#registering-a-node) | | Which user my image should run as (any of them — here is why) | [OPERATIONS.md](OPERATIONS.md#which-user-your-image-runs-as) | | Limits, quotas, and things that do not exist yet | [OPERATIONS.md](OPERATIONS.md#residual-limits-stated-plainly) | +| How to hand my node a file I uploaded in the orchestrator | [OPERATIONS.md](OPERATIONS.md#giving-the-node-a-file-you-uploaded) | +| Why my node started with no inputs, or has no Run button | [OPERATIONS.md](OPERATIONS.md#troubleshooting) | | Why my run is stuck at "Waiting for runner" | [OPERATIONS.md](OPERATIONS.md#troubleshooting) | --- @@ -133,9 +135,10 @@ reports it. | Capability | In the contract | Sent by the orchestrator today | Collected today | |---|---|---|---| -| Multiple input ports | Yes, `inputs` is a list with unique names | **No.** Exactly one port, always named `input` | not applicable | +| Multiple input ports | Yes, `inputs` is a list with unique names | **No.** At most one port, always named `input`; a job with nothing to read has none | not applicable | | Prefix (folder) input ports | Yes, `layout: "prefix"` with a tree digest | **No.** A folder artifact upstream is refused with a configuration error before the job is created | not applicable | -| Empty input set | Yes, `inputs` defaults to `[]` | **Yes.** A step with no upstream artifacts gets `"inputs": []` | not applicable | +| Empty input set | Yes, `inputs` defaults to `[]` | **Yes.** A step with nothing to read — no upstream artifacts, and no uploaded file on a Data Source wired in front of it — gets `"inputs": []` | not applicable | +| An operator's uploaded file as the input | Yes, an ordinary input object; no field was added to version 1 for it | **Yes.** A node wired directly after a Data Source node receives that node's uploaded file as one object on `input`, when the run carries no upstream artifacts. Uploaded files only, one file per run; see [PROTOCOL.md](PROTOCOL.md#22-invocationjson-field-by-field) | not applicable | | Multiple output ports | Yes, `produced_ports` is a name to relpath map | not applicable | **Yes.** Every port becomes a downstream artifact kind | | Objects claimed by no port | Yes | not applicable | Copied and verified, but **not offered downstream** on a successful run | | `result.json` as metrics | Yes, `ResultDoc` with `metrics` and `summary` | not applicable | **No. Nothing ever reads it.** Run metrics come from the marker inventory | diff --git a/tests/test_harness_self.py b/tests/test_harness_self.py index e4f4348..f2f47dd 100644 --- a/tests/test_harness_self.py +++ b/tests/test_harness_self.py @@ -511,9 +511,14 @@ def test_this_harness_agrees_with_the_orchestrators_own_frozen_documents(): a harness that shares code with the system it judges inherits its bugs. The price of that is drift, and this is what turns "they agree" from a claim into a check: the golden manifest carries a ``layout='prefix'`` port with a real ``prefix_digest``, an - ``organization_id``, ``upstream_execution_id`` on its objects and a ``null`` - ``prefix_digest`` on a file port — every one of which this harness's validator ignored - entirely until this round. + integer ``organization_id``, an integer ``upstream_execution_id`` on its objects and a + ``null`` ``prefix_digest`` on a file port — every one of which this harness's validator + ignored entirely until this round. + + One real form is NOT in these documents: ``upstream_execution_id`` as ``null``, which + is what an object carries when no producing execution is known for it. + ``test_this_copy_of_the_contract_accepts_an_input_whose_producing_execution_is_unknown`` + builds that one. """ manifest = json.loads((FIXTURES / 'golden_v1_manifest.json').read_bytes()) marker = json.loads((FIXTURES / 'golden_v1_marker.json').read_bytes()) @@ -557,6 +562,58 @@ def test_the_harness_writes_a_manifest_this_copy_of_the_contract_accepts(image, assert job.staging_prefix.endswith('attempts/4/gen-7/') +@harness_self_test +@traces_to( + 'external/contract.py declares the provenance field of InputObject as optional AND nullable — ' + '"upstream_execution_id: PositiveInt | None = None" — so a job description whose input objects carry ' + 'null there is a valid version-1 document. Its type says what else may stand there, only a real ' + 'integer of at least 1: "PositiveInt = Annotated[int, Field(strict=True, ge=1)]". A copy of the ' + 'contract that refused the null could not describe a job the orchestrator really sends; one that ' + 'stopped reading the field would accept documents the real parser refuses.' +) +def test_this_copy_of_the_contract_accepts_an_input_whose_producing_execution_is_unknown(): + """``null`` in the provenance field is a value, not a mistake. + + Setup: the orchestrator's golden manifest — the real parser's own reference + document — with ``upstream_execution_id`` set to ``null`` on every object. + That is the one real form of the field those documents do not contain: + all three of their objects carry the integer 89990. + Action: validate it with this harness's re-implementation of the contract; then + again with values that are neither ``null`` nor a real id. + Validate: the ``null`` form is accepted, and ``0``, ``true``, a numeric string and a + float are each refused. + + The refusals are what make the acceptance mean something. "Accepts ``null``" is also + exactly what a validator that never looks at the field does, so on its own it could + not tell this copy of the contract from one that had stopped checking provenance + altogether. + + ``null`` is not an exotic value. The launcher copies the id out of the run's own + trigger and writes ``null`` whenever there is none there — a run an operator started + by hand, for one. ``conformance/job.py`` can describe that job too, and without this + test a validator tightened to "must be an integer" would be noticed only as a + container test failing in ``setup()``, a long way from its cause. + """ + golden = (FIXTURES / 'golden_v1_manifest.json').read_bytes() + + def with_provenance(value) -> dict: + document = json.loads(golden) + objects = [obj for port in document['inputs'] for obj in port['objects']] + assert objects and all(obj['upstream_execution_id'] == 89990 for obj in objects), ( + 'the golden manifest no longer carries the integer form on every object; re-read the fixture ' + 'before trusting what this test says it adds' + ) + for obj in objects: + obj['upstream_execution_id'] = value + return document + + contract.validate_manifest(with_provenance(None)) + + for not_an_id in (0, True, '89990', 89990.0): + with pytest.raises(contract.ContractViolation): + contract.validate_manifest(with_provenance(not_an_id)) + + @harness_self_test @our_policy( 'The harness must inject what the agent injects and nothing it does not — above all not the legacy ' diff --git a/tests/test_inputs.py b/tests/test_inputs.py index 8004999..f335f36 100644 --- a/tests/test_inputs.py +++ b/tests/test_inputs.py @@ -23,6 +23,10 @@ from __future__ import annotations +import io +import json +import zipfile + import pytest from conformance import contract, docker @@ -110,6 +114,80 @@ def test_several_inputs_are_all_copied(make_job): assert job.endpoint.body_of(f'outputs/{spec.relpath}') == spec.data +def _a_real_zip_archive() -> bytes: + """A small, genuine zip archive: deterministic, and nothing like text. + + Built here rather than checked in, so the bytes are visibly what they claim to be, + and with a fixed timestamp so the same archive — and therefore the same pin — comes + out on every run. The member is STORED, not compressed, and holds every byte value + and a CR LF pair, so the archive itself contains NUL bytes, bytes that are not valid + UTF-8 and all three line endings: a step that decodes it as text, or translates a + line ending in either direction, cannot hand back the same bytes. + """ + buffer = io.BytesIO() + with zipfile.ZipFile(buffer, 'w', compression=zipfile.ZIP_STORED) as archive: + member = (bytes(range(256)) + b'\r\n') * 16 + archive.writestr(zipfile.ZipInfo('rows.bin', date_time=(2026, 1, 1, 0, 0, 0)), member) + return buffer.getvalue() + + +@conforms_today +@reference_quality( + _COPYING + + ' What this test adds is a SHAPE, not a new expectation. Every other input in this file is a line ' + 'or two of text or a block of generated filler, under a kind this harness invented, naming an ' + 'execution that produced it. A real job can be none of those things: ONE binary file on the one port ' + 'the orchestrator really builds ("input"), under a real kind ("zip" is what the platform\'s own Zip ' + 'step publishes), with "upstream_execution_id" null — which is what the launcher writes whenever ' + 'the run carries no upstream execution id, as on a run an operator started by hand. A step that ' + 'branched on the kind or required the provenance field would pass every other test in this file and ' + 'fail one of the plainest jobs there is; one that decoded its input as text would fail, in this ' + 'file, only this test and the oversized-input one.' +) +def test_one_binary_input_under_a_real_kind_with_no_producing_execution_is_copied(make_job): + """One archive, on one port, with nobody recorded as having produced it. + + Setup: ONE input object on the port named ``input`` — a genuine zip archive, so its + bytes are not text — with the port's ``payload_kind`` set to ``zip`` and the + object's ``upstream_execution_id`` ``null``. + Action: run. + Validate: the job description the store serves really has that shape; the step exits + 0 with a succeeded marker; and the archive is readable at + ``outputs/``, byte for byte. + + Everything asserted BEFORE the run is about this test's own fixture and this harness, + not about the node, and it is there on purpose. Without it the test would stay green + if the "archive" quietly became something a text decoder accepts, or if + ``conformance/job.py`` stopped writing the kind or the ``null`` and went back to + handing every container one invented shape. + """ + archive = _a_real_zip_archive() + assert zipfile.is_zipfile(io.BytesIO(archive)) + with pytest.raises(UnicodeDecodeError): + archive.decode('utf-8') + + job = make_job( + inputs=[InputSpec(relpath='batch.zip', data=archive)], + payload_kind='zip', + upstream_execution_id=None, + ) + served = json.loads(job.endpoint.blobs[contract.MANIFEST_FILENAME].data) + (port,) = served['inputs'] + (described,) = port['objects'] + assert (port['name'], port['payload_kind'], port['layout']) == ('input', 'zip', 'file'), port + assert 'upstream_execution_id' in described and described['upstream_execution_id'] is None, described + + result = job.run() + + assert result.exit_code == 0, result.output + assert job.marker()['status'] == 'succeeded' + copied = job.endpoint.body_of('outputs/batch.zip') + assert copied == archive, ( + f'the archive did not come through unchanged: {len(archive)} bytes went in and {len(copied)} came ' + f'out, {"the same length but different bytes" if len(copied) == len(archive) else "a different length"}' + ) + + @conforms_today @traces_to( 'pipelines/external_finalize.py _publish_and_verify resolves every inventoried relpath against the '