Added auto_write* methods to writers and clients.

[sbyrne-sf]

These new methods will be used with streaming transforms when running in specific modes where the system must control the write mode.

[jcatt-sf]

Automated code review (Claude Code / Sonnet) — findings ranked most severe first; inline comments anchor the specifics. Reproductions were run against this PR's head where noted.

🔴 Blocker

The new auto_write_* methods are on the wrong class, so the example this PR ships crashes. auto_write_to_dlo / auto_write_to_dmo are defined on StreamingClient (client.py:603/614), but streaming_deltas/entrypoint.py:57 calls auto_write_to_dlo on a plain Client() (line 54). get_run_mode() defaults to BATCH when BYOC_RUN_MODE is unset, so that else branch is the default path → AttributeError: 'Client' object has no attribute 'auto_write_to_dlo'. No test executes this template, so CI stays green. See inline on entrypoint.py:57 and client.py:603.

🟠 Medium

• Opaque error on a bad BYOC_RUN_MODE (client.py:636) — RunMode(run_mode) raises a bare ValueError that never names the env var; run_entrypoint runs user code via a bare runpy.run_path(), so it surfaces as a raw traceback.
• New @abstractmethods break the user-extensible writer contract (base.py:63) — any BaseDataCloudWriter subclass missing both methods becomes uninstantiable; the in-repo MockDataCloudWriter in tests/test_config.py wasn't updated (latent only because a test mocks subclass_from_config_name).
• Destructive, mode-blind default (csv.py:40) — the docstrings promise rebuild-vs-initial-sync behavior, but every auto_write_* hardcodes WriteMode.OVERWRITE and never calls get_run_mode(). For a DMO (governed, multi-source) an unconditional overwrite can discard other jobs' rows; every shipped example uses APPEND.

🟡 Low / polish

• Docstring contradicts control flow (entrypoint.py:20-22 vs the new else branch).
• source_dlo is dead on the DELTA_SYNC path (entrypoint.py:37).
• PrintDataCloudWriter duplicates bodies instead of delegating (print.py:125), unlike csv.py.
• Missing # type: ignore[no-any-return] (client.py:612/623) that every sibling return carries under warn_return_any = true; make lint runs mypy in CI (couldn't confirm locally).
• Two disconnected runtime-context mechanisms — get_run_mode() reads os.getenv directly, bypassing the config.json → ClientConfig path used for streaming_source; tests now need two idioms (reset_client fixture vs @patch.dict(os.environ)).
• auto_write_to_dmo has no real caller anywhere — even the new example only calls auto_write_to_dlo.
• Docs not updated — README.md and the template READMEs never mention auto_write_* or the BYOC_RUN_MODE / RunMode contract.

---

This review was generated by Claude Code from a diff review and posted on my behalf; please sanity-check before acting.

[jcatt-sf]

🟠 A typo'd BYOC_RUN_MODE (e.g. DETLA_SYNC) makes this raise ValueError: 'DETLA_SYNC' is not a valid RunMode — which never mentions BYOC_RUN_MODE at all. run_entrypoint executes user code via a bare runpy.run_path() with no surrounding try/except, so this reaches the top of a Spark job as an opaque traceback. Consider mirroring credentials.py's AuthType handling: catch ValueError and re-raise naming the env var and the valid modes.

[jcatt-sf]

🟡 source_dlo is only consumed by the else branch (read_dlo(source_dlo), line 55); the DELTA_SYNC branch calls read_dlo_deltas(), which takes no name argument. Minor, but a reader on the streaming path may expect it to be used — consider moving it into the else branch.