A Data Source's uploaded file as a node's input: docs, harness shape, two tests

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.

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,18 +171,32 @@ 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()
             port = ports.setdefault(
                 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,

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
 

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