@@ -205,8 +205,8 @@ def export_selected_files(files, output_path, *, fs, identity,
 
     Filesystem/authentication and retry wrappers stay with the caller. The
     directory ownership, output names and CSV retry decisions are shared.
-    ``sidecars`` (name to bytes) are written into a directory destination
-    after the data and before ``_SUCCESS``; ``fs`` then needs ``write_bytes``.
+    ``sidecars`` (name to bytes) land in a directory destination before
+    ``_SUCCESS``.
     """
     import os.path as osp
     import tempfile
 
@@ -57,11 +57,7 @@ def write_files(self, connector, table_name, files,
                     write_mode=WriteMode.OVERWRITE, project_id=None, *,
                     export_id=None, extra_columns=None,
                     sidecars: Mapping[str, bytes] = {}):
-        """Export selected Parquet without collecting their rows.
-
-        ``sidecars`` are written into a directory destination before
-        ``_SUCCESS``.
-        """
+        """Export selected Parquet without collecting their rows."""
         del project_id
         if write_mode is not WriteMode.OVERWRITE:
             raise NotImplementedError(
 
@@ -224,54 +224,33 @@ def interrupted(source, destination, **kwargs):
     assert (target / '_SUCCESS').exists()
 
 
-@pytest.mark.parametrize('directory', [True, False])
-def test_sidecars_are_written_before_success(tmp_path, monkeypatch,
-                                             directory):
-    import pyarrow.parquet as pq
-    from kumo_core.artifact_export import export_selected_files
-    from kumo_core.object_store_sink import _ArrowExportFiles
-    from pyarrow.fs import LocalFileSystem
-
-    source = tmp_path / 'part-0.parquet'
-    pq.write_table(pa.table({'id': [1]}), source)
-    target = tmp_path / 'output'
-    fs = _ArrowExportFiles(LocalFileSystem(), None)
-    touched = fs.touch
 
-    def touch(path):
-        assert (target / 'meta.json').read_bytes() == b'{}'
-        touched(path)
-
-    monkeypatch.setattr(fs, 'touch', touch)
-    export_selected_files([str(source)],
-                          str(target) + ('/' if directory else ''), fs=fs,
-                          identity='job-1', sidecars={'meta.json': b'{}'})
-    if directory:
-        assert (target / '_SUCCESS').exists()
-    else:
-        assert not (target / 'meta.json').exists()
-        assert pq.read_table(tmp_path / 'output.parquet').num_rows == 1
-
-
-def test_failed_sidecar_write_still_publishes(tmp_path):
+def test_sidecars_are_best_effort_and_precede_success(tmp_path):
     import pyarrow.parquet as pq
     from kumo_core.artifact_export import export_selected_files
     from kumo_core.object_store_sink import _ArrowExportFiles
     from pyarrow.fs import LocalFileSystem
 
     source = tmp_path / 'part-0.parquet'
     pq.write_table(pa.table({'id': [1]}), source)
+    target = tmp_path / 'output'
     fs = _ArrowExportFiles(LocalFileSystem(), None)
-    write_bytes = fs.write_bytes
+    write_bytes, touch = fs.write_bytes, fs.touch
 
     def denied(path, value):
-        if path.endswith('meta.json'):
+        if path.endswith('bad.json'):
             raise PermissionError('denied')
         write_bytes(path, value)
 
-    fs.write_bytes = denied
-    target = tmp_path / 'output'
+    def checked_touch(path):
+        assert (target / 'meta.json').read_bytes() == b'{}'
+        touch(path)
+
+    fs.write_bytes, fs.touch = denied, checked_touch
     with pytest.warns(UserWarning, match='denied'):
         export_selected_files([str(source)], f'{target}/', fs=fs,
-                              identity='job-1', sidecars={'meta.json': b'{}'})
+                              identity='job-1', sidecars={
+                                  'meta.json': b'{}',
+                                  'bad.json': b'{}'
+                              })
     assert (target / '_SUCCESS').exists()
@@ -18,8 +18,7 @@
 
 def stage_source_table(source_table: SourceTable,
                        destination: Path) -> Tuple[Path, Optional[str]]:
-    """Read ``source_table`` through its connector and retain its parquet
-    parts; also returns the connector's version of the rows read."""
+    """Stage ``source_table``'s parquet parts; returns them and their version."""
     if not isinstance(source_table, SourceTable):
         raise TypeError("source_table must be a kumo.engine SourceTable")
     destination.mkdir(parents=True, exist_ok=False)
 
@@ -2,27 +2,24 @@
 # All rights reserved.
 # SPDX-License-Identifier: Apache-2.0
 
-import json
-import logging
 import warnings
 from dataclasses import dataclass, replace
 from pathlib import Path
-from typing import TYPE_CHECKING, Any, Dict, Optional, Sequence
+from typing import TYPE_CHECKING, Any, Dict, Optional
 
 from kumo_core.model_plan import TrainingTableGenerationPlan
 from kumoapi.model_plan import \
     TrainingTableGenerationPlan as PublicGenerationPlan
 from kumoapi.train import TrainingTableSpec
 
+from kumo.engine import _provenance as provenance
 from kumo.engine._manifest import new_manifest, read_manifest, write_manifest
 from kumo.engine.connector.source_table import SourceTable
 from kumo.engine.graph import GraphSnapshot
 
 if TYPE_CHECKING:
     from kumo.engine.trainer.output_config import TrainingTableOutputConfig
 
-logger = logging.getLogger(__name__)
-
 STAGING_MANIFEST_FILE = 'training-table-manifest.json'
 EXPORT_MANIFEST_FILE = '_kumo_manifest.json'
 _RANDOM_SPLITS = ('RandomSplit', 'RandomLinkSplit')
@@ -84,40 +81,22 @@ def export(self, output_config: "TrainingTableOutputConfig") -> str:
         root.mkdir(parents=True, exist_ok=True)
         with tempfile.TemporaryDirectory(dir=root) as output:
             files = export_train_table(str(self.staging_dir), output)
-            try:
-                payload = self._export_manifest(connector, output_config,
-                                                files)
-            except Exception as error:
-                warnings.warn(f"Exporting without a training-table manifest: "
-                              f"{error}", UserWarning, stacklevel=2)
-                payload = None
-            extra = ({} if payload is None else {
-                'sidecars': {
-                    EXPORT_MANIFEST_FILE: payload
-                }
-            })
             return write_files(connector, output_config.table_name, files,
                                output_config.write_mode, export_id=export_id,
-                               **extra)
+                               **self._export_sidecars(connector,
+                                                       output_config))
 
-    def _export_manifest(self, connector, output_config,
-                         files: Sequence[str]) -> Optional[bytes]:
+    def _export_sidecars(self, connector, output_config) -> Dict[str, Any]:
         from kumo_core.object_store_sink import ObjectStoreSinkBackend
         from kumo_core.sources import layout
 
-        name = output_config.table_name
+        manifest = self.staging_dir / STAGING_MANIFEST_FILE
+        name = output_config.table_name.rstrip('/')
         if (not isinstance(connector.sink_backend, ObjectStoreSinkBackend)
-                or not (name.endswith('/') or len(files) > 1)
-                or layout.file_type(name.rstrip('/')) not in (
-                    None, layout.PARQUET)):
-            logger.debug('Training-table manifests are only exported to '
-                         'Parquet directories on files or object stores.')
-            return None
-        manifest = self.manifest
-        if manifest is None:
-            return None
-        return (json.dumps(manifest, indent=2, sort_keys=True) +
-                '\n').encode('utf-8')
+                or layout.file_type(name) == layout.CSV
+                or not manifest.is_file()):
+            return {}
+        return {'sidecars': {EXPORT_MANIFEST_FILE: manifest.read_bytes()}}
 
     @classmethod
     def from_source_table(
@@ -182,7 +161,6 @@ def from_source_table(
         run_dir = train_table_scratch_dir(run_id)
         rows_path, version = stage_source_table(source_table,
                                                 run_dir / "source-table")
-        lineage = _source_lineage(source_table, version)
         generation_plan = from_json_dict(
             to_json_dict(public_plan),
             TrainingTableGenerationPlan,
@@ -247,8 +225,11 @@ def from_source_table(
             required_stages=task_plan.task_dataset.required_stages,
             generation_plan=generation_plan,
         )
-        _record_manifest(table, snapshot, task_plan.train_table, 'custom',
-                         source_table=lineage)
+        _record_manifest(
+            table, snapshot, task_plan.train_table, 'custom',
+            source_table=provenance.source_table(source_table.connector,
+                                                 source_table.table_name,
+                                                 source_table.sql, version))
         return table
 
 
@@ -285,11 +266,3 @@ def _record_manifest(table: TrainingTable, snapshot: GraphSnapshot,
         warnings.warn(
             f"Could not write the training-table manifest at {path}: "
             f"{error}", UserWarning, stacklevel=3)
-
-
-def _source_lineage(source_table: SourceTable,
-                    version: Optional[str]) -> Dict[str, Optional[str]]:
-    from kumo.engine._provenance import source_table as lineage
-
-    return lineage(source_table.connector, source_table.table_name,
-                   source_table.sql, version)
@@ -295,84 +295,38 @@ def test_custom_training_table_declares_integer_timestamp_unit(graph, unit,
         assert _rows(seeds[stage], columns) == _rows(original_seeds[stage], columns)
 
 
-def _reload(graph, name, pquery):
-    _, snapshot, connector, _ = graph
-    return kumo.TrainingTable.from_source_table(
-        connector[name], TrainingTableSpec(split_col='SPLIT'), pq=pquery,
-        snapshot=snapshot)
-
-
 def test_manifest_and_export_sidecar(graph, tmp_path):
     from kumo_core.training.export import export_train_table
 
     _, snapshot, connector, out = graph
     query = 'PREDICT ents.LABEL FOR EACH ents.ID'
     pquery, table, _ = _export(graph, query, 'lineage')
     manifest = table.manifest
-    assert manifest['kind'] == 'training_table'
     assert manifest['id'] == table.tt_id == _spec(table)['tt_id']
     assert manifest['parents'] == {'snapshot': snapshot.id}
     assert (manifest['query'], manifest['source']) == (query, 'pql')
-    assert manifest['task_type'] == 'binary_classification'
     assert manifest['rows'] == _seeds(table)[1]
     assert manifest['split_seed'] == 0
-    assert manifest['generation_plan']['split'] == 'inferred'
 
     directory = out / 'lineage'
-    parts = sorted(directory.glob('*.parquet'))
     baseline = tmp_path / 'baseline'
-    expected = export_train_table(str(table.staging_dir), str(baseline))
-    assert [p.name for p in parts] == [p.split('/')[-1] for p in expected]
-    assert [p.read_bytes() for p in parts] == [
-        (baseline / p.name).read_bytes() for p in parts
-    ]
-    assert sorted(p.name for p in directory.iterdir()) == sorted(
-        [p.name for p in parts] +
-        ['_KUMO_EXPORT', '_SUCCESS', '_kumo_manifest.json'])
-    sidecar = (directory / '_kumo_manifest.json').read_bytes()
-    assert json.loads(sidecar) == manifest
-
-    table.export(TrainingTableOutputConfig(connector, 'lineage'))
-    assert (directory / '_kumo_manifest.json').read_bytes() == sidecar
-
-
-def test_custom_table_records_its_source(graph, recwarn):
-    _, snapshot, connector, out = graph
-    pquery, table, _ = _export(graph, 'PREDICT ents.LABEL FOR EACH ents.ID',
-                               'reloaded')
-    (out / 'reloaded' / '_kumo_manifest.json').write_text('{not json')
-    reloaded = _reload(graph, 'reloaded', pquery)
+    export_train_table(str(table.staging_dir), str(baseline))
+    for part in baseline.glob('*.parquet'):
+        assert (directory / part.name).read_bytes() == part.read_bytes()
+    assert json.loads(
+        (directory / '_kumo_manifest.json').read_bytes()) == manifest
+
+    reloaded = kumo.TrainingTable.from_source_table(
+        connector['lineage'], TrainingTableSpec(split_col='SPLIT'), pq=pquery,
+        snapshot=snapshot)
     manifest = reloaded.manifest
     assert manifest['source'] == 'custom'
-    assert manifest['id'] != table.tt_id
-    assert manifest['parents'] == {
-        'snapshot': snapshot.id,
-        'source_table': {
-            'connector': 'LOCAL_FILE',
-            'location': str(out / 'reloaded'),
-            'version': connector.source_backend.get_data_version(
-                connector, 'reloaded'),
-        },
+    assert manifest['parents']['source_table'] == {
+        'connector': 'LOCAL_FILE',
+        'location': str(directory),
+        'version': connector.source_backend.get_data_version(
+            connector, 'lineage'),
     }
-    assert manifest['parents']['source_table']['version']
-    assert not [w for w in recwarn if 'manifest' in str(w.message)]
-    assert _seeds(reloaded)[1] == _seeds(table)[1]
-
-
-def test_unrecordable_location_is_null(graph, monkeypatch):
-    from kumo.engine import _provenance
-
-    pquery, table, _ = _export(graph, 'PREDICT ents.LABEL FOR EACH ents.ID',
-                               'unrecorded')
-
-    def fail(*args, **kwargs):
-        raise OSError('listing denied')
-
-    monkeypatch.setattr(_provenance, 'source_location', fail)
-    with pytest.warns(UserWarning, match='listing denied'):
-        reloaded = _reload(graph, 'unrecorded', pquery)
-    recorded = reloaded.manifest['parents']['source_table']
-    assert recorded['location'] is None and recorded['version']
     assert _seeds(reloaded)[1] == _seeds(table)[1]
 
 
@@ -387,37 +341,27 @@ def fail(*args, **kwargs):
 
     monkeypatch.setattr(training_table, 'write_manifest', fail)
     with pytest.warns(UserWarning, match='read-only scratch'):
-        pquery, table, exported = _export(
+        _, table, exported = _export(
             graph, 'PREDICT ents.LABEL FOR EACH ents.ID', 'no_manifest')
     assert table.manifest is None
     assert not (out / 'no_manifest' / '_kumo_manifest.json').exists()
-    assert (out / 'no_manifest' / '_SUCCESS').exists()
     assert exported.num_rows == sum(_seeds(table)[1].values())
 
 
-def test_warehouse_sink_receives_no_sidecars(graph):
-    graph_, snapshot, _, _ = graph
+def test_no_sidecar_for_warehouse_or_csv_exports(graph):
+    graph_, snapshot, connector, out = graph
     table = kumo.PredictiveQuery(
         graph=graph_, query='PREDICT ents.LABEL FOR EACH ents.ID'
     ).generate_training_table(snapshot)
     calls = []
 
     def write_files(connector, table_name, files, write_mode, **kwargs):
         calls.append(kwargs)
-        return table_name
 
-    connector = SimpleNamespace(
+    warehouse = SimpleNamespace(
         source_type='SNOWFLAKE',
         sink_backend=SimpleNamespace(write_files=write_files))
-    table.export(TrainingTableOutputConfig(connector, 'DB.SCHEMA.TT'))
+    table.export(TrainingTableOutputConfig(warehouse, 'DB.SCHEMA.TT'))
     assert calls == [{'export_id': f'{table.tt_id}:tt-export-v1'}]
-
-
-def test_csv_directory_export_gets_no_sidecar(graph):
-    graph_, snapshot, connector, out = graph
-    table = kumo.PredictiveQuery(
-        graph=graph_, query='PREDICT ents.LABEL FOR EACH ents.ID'
-    ).generate_training_table(snapshot)
     table.export(TrainingTableOutputConfig(connector, 'rows.csv/'))
-    names = {p.name for p in (out / 'rows.csv').iterdir()}
-    assert '_SUCCESS' in names and '_kumo_manifest.json' not in names
+    assert not (out / 'rows.csv' / '_kumo_manifest.json').exists()