diff --git a/CHANGELOG.md b/CHANGELOG.md
index 80355f9..180314b 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -11,6 +11,28 @@ there was nothing to have changed from. Everything above it is ordinary.
 
 ### Added
 
+- **`Stream.adopt(name, root=, published=, schema=, end_offset=)`**: continue
+  a stream whose only remaining copy is its published table. The third way a
+  stream arrives on a box: `new` refuses a location that already holds a
+  table, and `restore` rebuilds from a replicated WAL that a producer with
+  `wal_replication` off never shipped. `adopt` is for that producer's table
+  once the producer is stopped for good.
+  - **`migrate`'s shape at a seam nobody sealed:** the table joins the
+    stream's metadata as a retired log, `[start, end)` read from the bucket,
+    and a new log (`<name>-v2`, then `-v3`, …) is created at exactly `end`,
+    so offsets stay one dense sequence and nothing the old producer issued
+    is reissued. Snapshots and `connect(catch_up=True)` read both sides; the
+    server replays the new log and refuses a cursor below the seam with
+    `evicted`. A stream already migrated once is adopted from its published
+    metadata.
+  - **`end_offset`** is what the old producer said its log ended at; a table
+    that disagrees is published short (or long) and nothing is adopted.
+  - **Refused** when the box already holds the stream (`migrate` opens it),
+    when nothing is published there, and when a column's type differs from
+    what the table had.
+  - **No manifest** for the adopted table: its statistics were never read on
+    this box, so a snapshot opens it unpruned, as it does a live log.
+
 - **`Subscription.recv_many(limit=500)` and `Subscription.batches(limit=500)`**
   (#86): a subscriber's version of group commit.
   - **What a batch holds:** at least one row, then every row already on the
diff --git a/src/streamcast/_stream.py b/src/streamcast/_stream.py
index 09e9eec..3649859 100644
--- a/src/streamcast/_stream.py
+++ b/src/streamcast/_stream.py
@@ -39,11 +39,14 @@
 import contextlib
 import dataclasses
 import time
+import uuid
 from functools import partial
 from pathlib import Path
 from typing import TYPE_CHECKING, Final
 
+import duckdb
 import litelink
+import pyarrow as pa
 from websockets.exceptions import ConnectionClosed
 from websockets.frames import CloseCode
 
@@ -53,6 +56,7 @@
     _log,
     _manifest,
     _metadata,
+    _published,
     _replicate,
     _schema,
     _snapshot,
@@ -694,6 +698,166 @@ def restore(
             group_commit=group_commit,
         )
 
+    @classmethod
+    def adopt(
+        cls,
+        name: str = "",
+        *,
+        root: str | PathLike[str],
+        published: str,
+        schema: Mapping[str, object],
+        sort_by: Sequence[str] | None = None,
+        config: object | None = None,
+        s3_options: object | None = None,
+        end_offset: int | None = None,
+        max_replay: int | None = MAX_REPLAY,
+        replay_published: bool = False,
+        group_commit: bool = True,
+    ) -> Stream:
+        """Continue a stream whose only remaining copy is its published table.
+
+            stream = streamcast.Stream.adopt(
+                "trades", root="data", published="s3://market-data/prod", schema=V2
+            )
+
+        The third way a stream arrives on a box, beside `new` and `restore`.
+        `new` refuses a published location that already holds a table, and
+        `restore` rebuilds the log from a replicated WAL — which a producer
+        that ran with `wal_replication` off never shipped. This is for that
+        producer's table once the producer is gone for good: the table is the
+        whole history, published in full, and the stream carries on from it.
+
+        **What it does, in `migrate`'s shape.** The published table joins the
+        stream's metadata as a RETIRED log, exactly as a log `migrate` sealed
+        would: `[start, end)` are its own offsets, read from the bucket. A new
+        log, `<name>-v2` (then `-v3`, …), is created at EXACTLY `end`, so the
+        offsets stay one dense sequence across the seam and no offset the old
+        producer issued is ever reissued. The metadata is written beside the
+        tables, and the stream is returned ready to `serve` and `send` to.
+        Snapshots and `connect(catch_up=True)` read the retired table through
+        the metadata like any retired log; the server itself replays only the
+        new one, and refuses a cursor below the seam with `evicted` naming it.
+
+        **No manifest for the retired table**, since its statistics were never
+        read on this box: a snapshot opens it unpruned, as it opens a live log.
+
+        `end_offset` is what the old producer said its log ended at, if it said
+        — one past its last row — and the table has to agree, or the producer
+        published short and NOTHING is adopted: the rows it holds would be lost
+        at the seam, and a short table is the one thing nobody notices later.
+        A stream already migrated once is adopted from its published metadata:
+        its current log's table is sealed at its end and the next name follows.
+
+        ⚠️ **Stop the old producer first, for good.** Nothing here can tell
+        that it is still alive, and two writers on one published table is the
+        corruption `restore` warns of. Where there is a replica, `restore`
+        keeps the rows a planned stop left unpublished; this keeps only what
+        the table holds, which after a full publish is everything.
+
+        Refused — `FileExistsError` — when this box already has the stream
+        (its metadata, or a log at its name): `migrate` opens that.
+        """
+        declared = _declaration(schema)
+        if _metadata.load(root, name) is not None or Path(root, name).exists():
+            msg = (
+                f"stream {name!r} is already at {root}; Stream.migrate opens it. "
+                f"adopt is for a box that never held the stream"
+            )
+            raise FileExistsError(msg)
+
+        # The published copy of the metadata, which only a remote location has:
+        # a `file://` one keeps it beside the log, on the box that is gone.
+        metadata = (
+            _metadata.fetch(published, name, s3_options)  # ty: ignore[invalid-argument-type]
+            if _metadata.remote(published)
+            else None
+        )
+        current = name if metadata is None else metadata.current.name
+        location = (
+            published if metadata is None else (metadata.current.published or published)
+        )
+        try:
+            table = _published.Table.open(location, current, s3_options)  # ty: ignore[invalid-argument-type]
+        except (FileNotFoundError, duckdb.IOException) as exc:
+            # No `version-hint.text` there: nothing was ever published at this
+            # location, or the prefix is wrong. Either way there is no table.
+            msg = (
+                f"no published table at {location}/{current}, so there is nothing "
+                f"to adopt ({exc}). Stream.new creates a stream; Stream.restore "
+                f"rebuilds one from its replica"
+            )
+            raise FileNotFoundError(msg) from exc
+        try:
+            extent = table.extent
+            held = pa.schema(
+                [field for field in table.schema if field.name != _log.COLUMN]
+            )
+        finally:
+            table.close()
+        if extent is None:
+            msg = (
+                f"{location}/{current} holds no rows, so there is nothing to adopt. "
+                f"Stream.new creates a stream; Stream.restore rebuilds one from its "
+                f"replica"
+            )
+            raise FileNotFoundError(msg)
+        start, end = extent
+        if end_offset is not None and end_offset != end:
+            short = end < end_offset
+            msg = (
+                f"{location}/{current} ends at offset {end} and the old producer said "
+                f"{end_offset}: the table is published "
+                f"{'short' if short else 'past what the producer reported'}. "
+                f"Publish it in full and adopt again; nothing was adopted"
+            )
+            raise ValueError(msg)
+
+        if metadata is None:
+            metadata = _metadata.Metadata(
+                stream=name,
+                stream_id=str(uuid.uuid4()),
+                sealed_logs=(),
+                live_log=_metadata.describe(
+                    name, start, None, held, published=published
+                ),
+            )
+        _metadata.check_types(metadata.logs, _schema.from_arrow(declared))
+
+        new_log = litelink.new(
+            root,
+            metadata.next_name(),
+            schema=_log.with_system(declared),
+            sort_by=sort_by,
+            config=config,  # ty: ignore[invalid-argument-type]
+            published=published,
+            s3_options=s3_options,  # ty: ignore[invalid-argument-type]
+            start_offset=end,
+        )
+        metadata = metadata.advance(
+            end,
+            _metadata.describe(
+                new_log.name, end, None, new_log.schema, published=new_log.published
+            ),
+        )
+        try:
+            _metadata.save(root, metadata)
+            if _metadata.remote(new_log.published):
+                _metadata.publish(metadata, new_log.published, s3_options)  # ty: ignore[invalid-argument-type]
+        except BaseException:
+            new_log.close()
+            raise
+        return cls(
+            name,
+            log=new_log,
+            owns_log=True,
+            max_replay=max_replay,
+            floor=_floor(metadata),
+            retired=_retired(root, metadata),
+            s3_options=s3_options,  # ty: ignore[invalid-argument-type]
+            replay_published=replay_published,
+            group_commit=group_commit,
+        )
+
     def __repr__(self) -> str:
         durable = "durable" if self._log is not None else "live-only"
         return (
diff --git a/tests/test_adopt.py b/tests/test_adopt.py
new file mode 100644
index 0000000..c939a6d
--- /dev/null
+++ b/tests/test_adopt.py
@@ -0,0 +1,258 @@
+"""`Stream.adopt`: a stream continues from a published table nobody holds.
+
+The case neither `new` nor `restore` covers: the old producer ran with
+`wal_replication` off, published its log in full, and is gone. `new` refuses
+the location (it holds a table) and `restore` has no replica to rebuild from.
+What matters here is `migrate`'s properties at a seam nobody sealed: the
+offsets stay one dense sequence, nothing the old producer issued is reissued,
+and every reader of the stream sees both sides.
+"""
+
+from __future__ import annotations
+
+import shutil
+from pathlib import Path
+
+import pytest
+
+import streamcast
+from streamcast import _metadata
+from tests.test_migrate import V1, V2, evolve, row
+
+
+async def _old_producer(root: Path, published: str, s3=None, count: int = 5) -> None:
+    """A producer that published everything it had and was then taken down."""
+    stream = streamcast.Stream.new(
+        "trades", root=root, schema=V1, published=published, s3_options=s3
+    )
+    try:
+        await stream.send_many([row(i) for i in range(count)])
+        assert stream.log is not None
+        stream.log.seal(flush=True)
+        stream.log.publish(flush=True)
+        assert stream.log.published_through() == count
+    finally:
+        await stream.aclose()
+    shutil.rmtree(root)  # the box is gone; the table is all that is left
+
+
+class TestItContinuesTheTable:
+    async def test_the_new_log_starts_where_the_table_ends(self, tmp_path):
+        published = (tmp_path / "bucket").as_uri()
+        await _old_producer(tmp_path / "box_a", published)
+
+        stream = streamcast.Stream.adopt(
+            "trades", root=tmp_path / "box_b", published=published, schema=V2
+        )
+        try:
+            assert stream.log is not None
+            assert stream.log.name == "trades-v2"
+            assert stream.log.end_offset() == 6, "one past the table's last row"
+            assert stream.name == "trades"
+            # The two rows sent now take the offsets after the seam, densely.
+            assert await stream.send_many([row(5, venue="x"), row(6, venue="y")]) == [
+                6,
+                7,
+            ]
+        finally:
+            await stream.aclose()
+
+        metadata = _metadata.load(tmp_path / "box_b", "trades")
+        assert metadata is not None
+        (old,) = metadata.sealed_logs
+        assert (old.name, old.start_offset, old.end_offset, old.published) == (
+            "trades",
+            1,
+            6,
+            published,
+        )
+        assert set(old.schema["properties"]) == {"event_ts", "price", "side"}
+        assert metadata.live_log.name == "trades-v2"
+        assert metadata.live_log.start_offset == 6
+        assert "venue" in metadata.live_log.schema["properties"]
+        assert metadata.manifest is None
+        # (A `file://` location gets no published copy of the metadata; the S3
+        # test below checks that one.)
+
+    async def test_a_snapshot_reads_both_sides_of_the_seam(self, tmp_path):
+        published = (tmp_path / "bucket").as_uri()
+        await _old_producer(tmp_path / "box_a", published)
+        stream = streamcast.Stream.adopt(
+            "trades", root=tmp_path / "box_b", published=published, schema=V2
+        )
+        try:
+            await stream.send_many([row(5, venue="x"), row(6, venue="y")])
+            stream.log.seal(flush=True)
+            stream.log.publish(flush=True)
+            stream.ensure_metadata()
+            uri = stream.metadata_uri
+            assert uri is not None
+        finally:
+            await stream.aclose()
+
+        async with await streamcast.Stream.snapshot(uri) as snap:
+            table = await snap.sql(
+                "SELECT event_ts, price, venue FROM log ORDER BY event_ts"
+            )
+        assert table.num_rows == 7
+        assert table.column("event_ts").to_pylist() == [
+            1_790_000_000_000_000 + i for i in range(7)
+        ]
+        # The old table never had `venue`: null there, filled on the new side.
+        assert table.column("venue").to_pylist() == [None] * 5 + ["x", "y"]
+
+    async def test_a_subscriber_catching_up_from_the_start_gets_every_row(
+        self, tmp_path, serve
+    ):
+        published = (tmp_path / "bucket").as_uri()
+        await _old_producer(tmp_path / "box_a", published)
+        stream = streamcast.Stream.adopt(
+            "trades", root=tmp_path / "box_b", published=published, schema=V2
+        )
+        async with serve(stream) as uri:
+            await stream.send_many([row(5, venue="x"), row(6, venue="y")])
+            async with streamcast.connect(uri, offset=1, catch_up=True) as sub:
+                got = [await sub.recv() for _ in range(7)]
+        assert [offset for offset, _ts, _row in got] == list(range(1, 8))
+        assert [r["price"] for _o, _t, r in got] == [100.0 + i for i in range(7)]
+
+    async def test_it_is_served_again_by_migrate_at_the_next_start(self, tmp_path):
+        """The server's startup path is `migrate` (idempotent): an adopted stream
+        is one it opens as-is, at its current log, with the seam intact."""
+        published = (tmp_path / "bucket").as_uri()
+        await _old_producer(tmp_path / "box_a", published)
+        stream = streamcast.Stream.adopt(
+            "trades", root=tmp_path / "box_b", published=published, schema=V2
+        )
+        await stream.aclose()
+        again = streamcast.Stream.migrate("trades", root=tmp_path / "box_b", schema=V2)
+        try:
+            assert again.log is not None and again.log.name == "trades-v2"
+            assert again.log.end_offset() == 6
+        finally:
+            await again.aclose()
+
+
+class TestItRefuses:
+    async def test_a_table_published_short_of_what_the_producer_said(self, tmp_path):
+        published = (tmp_path / "bucket").as_uri()
+        await _old_producer(tmp_path / "box_a", published)
+        with pytest.raises(ValueError, match="published short"):
+            streamcast.Stream.adopt(
+                "trades",
+                root=tmp_path / "box_b",
+                published=published,
+                schema=V2,
+                end_offset=9,
+            )
+        assert not (tmp_path / "box_b" / "trades-v2").exists(), "nothing was adopted"
+        assert _metadata.load(tmp_path / "box_b", "trades") is None
+        # And exactly right is accepted.
+        stream = streamcast.Stream.adopt(
+            "trades",
+            root=tmp_path / "box_b",
+            published=published,
+            schema=V2,
+            end_offset=6,
+        )
+        await stream.aclose()
+
+    async def test_a_location_with_nothing_published(self, tmp_path):
+        (
+            tmp_path / "bucket"
+        ).mkdir()  # the prefix exists; no table was ever published under it
+        published = (tmp_path / "bucket").as_uri()
+        with pytest.raises(FileNotFoundError, match="nothing to adopt"):
+            streamcast.Stream.adopt(
+                "trades", root=tmp_path / "box_b", published=published, schema=V2
+            )
+
+    async def test_a_box_that_already_holds_the_stream(self, tmp_path):
+        published = (tmp_path / "bucket").as_uri()
+        await _old_producer(tmp_path / "box_a", published)
+        stream = streamcast.Stream.adopt(
+            "trades", root=tmp_path / "box_b", published=published, schema=V2
+        )
+        await stream.aclose()
+        with pytest.raises(FileExistsError, match="Stream.migrate opens it"):
+            streamcast.Stream.adopt(
+                "trades", root=tmp_path / "box_b", published=published, schema=V2
+            )
+
+    async def test_a_column_whose_type_changed(self, tmp_path):
+        published = (tmp_path / "bucket").as_uri()
+        await _old_producer(tmp_path / "box_a", published)
+        retyped = evolve(V1, price={"type": "integer"})
+        with pytest.raises(ValueError, match="fixed for the life of a stream"):
+            streamcast.Stream.adopt(
+                "trades", root=tmp_path / "box_b", published=published, schema=retyped
+            )
+        assert not (tmp_path / "box_b" / "trades-v2").exists()
+
+
+@pytest.mark.replication
+class TestOnObjectStorage:
+    async def test_the_same_round_trip_against_s3(self, tmp_path, s3, bucket, serve):
+        await _old_producer(tmp_path / "box_a", bucket, s3)
+        stream = streamcast.Stream.adopt(
+            "trades",
+            root=tmp_path / "box_b",
+            published=bucket,
+            schema=V2,
+            s3_options=s3,
+            end_offset=6,
+        )
+        async with serve(stream) as uri:
+            assert stream.log is not None and stream.log.end_offset() == 6
+            await stream.send_many([row(5, venue="x")])
+            async with streamcast.connect(
+                uri, offset=1, catch_up=True, s3_options=s3
+            ) as sub:
+                got = [await sub.recv() for _ in range(6)]
+        assert [offset for offset, _ts, _row in got] == list(range(1, 7))
+        assert _metadata.fetch(bucket, "trades", s3) == _metadata.load(
+            tmp_path / "box_b", "trades"
+        ), "published beside the tables, the same file"
+
+    async def test_a_stream_migrated_once_is_adopted_from_its_metadata(
+        self, tmp_path, s3, bucket
+    ):
+        """Adopted, served, lost again: the published metadata names the current
+        log, whose table is sealed at its end, and the next name follows."""
+        await _old_producer(tmp_path / "box_a", bucket, s3)
+        stream = streamcast.Stream.adopt(
+            "trades",
+            root=tmp_path / "box_b",
+            published=bucket,
+            schema=V2,
+            s3_options=s3,
+        )
+        try:
+            await stream.send_many([row(5, venue="x"), row(6, venue="y")])
+            stream.log.seal(flush=True)
+            stream.log.publish(flush=True)
+        finally:
+            await stream.aclose()
+        shutil.rmtree(tmp_path / "box_b")
+
+        V3 = evolve(V2, note={"type": ["string", "null"]})
+        third = streamcast.Stream.adopt(
+            "trades",
+            root=tmp_path / "box_c",
+            published=bucket,
+            schema=V3,
+            s3_options=s3,
+        )
+        try:
+            assert third.log is not None and third.log.name == "trades-v3"
+            assert third.log.end_offset() == 8
+            metadata = _metadata.load(tmp_path / "box_c", "trades")
+            assert [
+                (e.name, e.start_offset, e.end_offset) for e in metadata.sealed_logs
+            ] == [
+                ("trades", 1, 6),
+                ("trades-v2", 6, 8),
+            ]
+            assert _metadata.fetch(bucket, "trades", s3) == metadata
+        finally:
+            await third.aclose()