diff --git a/.drive/projects/prisma-cli-v8/assets/briefs/deployment-logs-http-endpoint.md b/.drive/projects/prisma-cli-v8/assets/briefs/deployment-logs-http-endpoint.md
new file mode 100644
index 00000000..84553d50
--- /dev/null
+++ b/.drive/projects/prisma-cli-v8/assets/briefs/deployment-logs-http-endpoint.md
@@ -0,0 +1,113 @@
+# Brief: serve deployment logs over HTTP (for the platform/control-plane team)
+
+Written 2026-08-13, for an agent working in `pdp-control-plane` with
+no prior context on the CLI project. Operator: Will Madden.
+
+## The ask, in one paragraph
+
+Add an HTTP variant of the deployment-logs endpoint: the same
+log/terminal records `GET /v1/deployments/{deploymentId}/logs`
+streams over its WebSocket today, served instead as newline-delimited
+JSON over a plain authenticated `GET`, keeping the `cursor` resume
+semantics and the in-band terminal record. This was agreed in
+principle on 2026-08-12 (team answer, via Will): **"HTTP instead of
+WebSockets would be acceptable, as long as we can add live streaming
+at a later date."** This brief is the concrete version of that ask.
+
+## Why the CLI needs it
+
+The v8 Prisma CLI (repo `prisma/prisma-cli`, built on
+`@prisma/cli-engine`) must ship a `service deployment logs` command —
+the last unported command in its entire surface. The old
+implementation reached around the engine: it took a raw token and let
+`@prisma/compute-sdk`'s `streamLogs` build a `wss:` URL and set the
+Authorization header itself. The CLI's credential model now forbids
+that — credentials never reach command code; the engine holds them —
+so the command was shelved rather than shipped
+(see the CLI-side design doc below).
+
+The CLI's engine speaks HTTP only. Its `build logs` command already
+consumes the sibling build-logs endpoint as a streamed HTTP response
+(`packages/cli/src/v8/build/logs.ts` in prisma-cli — openapi-fetch,
+`parseAs: "stream"`, NDJSON lines). If deployment logs serves the
+same shape, the CLI needs zero new transport machinery and the
+engine's WebSocket affordance stays unbuilt. That design exists and
+is deliberately shelved as the future live-streaming path:
+
+- **CLI transport design (read §5 and §7):**
+  <https://github.com/prisma/prisma-cli/blob/main/.drive/projects/prisma-cli-v8/assets/engine/websocket-transport-design.md>
+  §7's question 2 ("could this be plain HTTP?") is the question your
+  team answered yes to; §5 records why reconnection-across-segments
+  must survive in whatever transport ships.
+
+## What exists on your side (verified against the repo, HEAD `deb158e4e`)
+
+- **The route** is WebSocket-only. Spec text in
+  `packages/management-api-sdk/src/api.d.ts` under
+  `"/v1/deployments/{deploymentId}/logs"`: upgrades to a socket;
+  messages are `type: "log"` (text + byte metadata) or
+  `type: "terminal"` (end-of-segment with reconnect cursor); the
+  stream ends after 10 minutes; reconnect with `cursor`.
+- **The interactor** `packages/interactors/src/compute/streamLogs.ts`
+  is a poll relay, not a push source: it polls Foundry's VM logs
+  every `DEFAULT_POLL_INTERVAL_MS = 1_000`, chunks tails at
+  `TAIL_CHUNK_SIZE = 10_240` (a Unikraft limit, per its comment),
+  runs `SEGMENT_DURATION_MS = 10 * 60 * 1_000` segments, defaults to
+  `DEFAULT_TAIL_LINES = 100`, holds a lease via
+  `computeLogStreamLease.repository.ts`, and maps Foundry's 424 (no
+  VM assigned / deallocated) explicitly. Record shapes:
+  `LogLine { type: "log"; text; byteStart; byteEnd }` and
+  `TerminalLine { type: "terminal"; kind: "end" | "error"; code;
+  message; retryable; cursor; details? }`.
+  Architecture: `docs/architecture/adrs/ADR-002-compute-log-relay-architecture.md`.
+- **The template already in your repo**:
+  `packages/interactors/src/compute/streamBuildLogs.ts` feeds the
+  build-logs endpoint the CLI consumes over plain HTTP today. The
+  deployment-logs HTTP variant is that shape fed by `streamLogs`.
+
+Because the source is a 1-second poll relay, serving it over a
+streamed HTTP response loses nothing real — there is no push
+latency to preserve. Genuine live streaming stays a later upgrade,
+which is exactly what the 2026-08-12 answer reserved.
+
+## Contract the CLI will consume (pin this before shipping)
+
+1. Authenticated `GET` (Authorization header; same credential as the
+   rest of the management API — no token in the URL, ever).
+2. Response: newline-delimited JSON; each line one record with the
+   EXISTING shapes — `type: "log"` and `type: "terminal"` unchanged.
+   The CLI maps `terminal.kind`/`retryable` onto its settlement, so
+   the terminal record must arrive in-band, including on the routine
+   10-minute segment end (`kind: "end"` with a `cursor`).
+3. `cursor` query parameter resumes a segment chain; a `tail`
+   parameter for initial history if the WS contract exposes one
+   (interactor default is 100 lines).
+4. The endpoint lands in the management-api OpenAPI spec so the
+   generated SDK (`@prisma/management-api-sdk`) exposes it — the CLI
+   consumes it through that SDK's types, not a hand-built URL.
+   Whether it is a new path or content negotiation on the existing
+   one is your call; the CLI only needs it addressable through the
+   generated client.
+5. Still marked experimental is fine. Tell the CLI project when the
+   contract is pinned and when it deploys — that unshelves the
+   command.
+
+## CLI-side context, for pointers rather than action
+
+- **Project plan** (S8 section records the whole history of this):
+  <https://github.com/prisma/prisma-cli/blob/main/.drive/projects/prisma-cli-v8/plan.md>
+- **The slice contract that ruled logs out of the last slice**
+  (R-S8-5 records the 2026-08-12 answer verbatim):
+  <https://github.com/prisma/prisma-cli/blob/main/.drive/projects/prisma-cli-v8/specs/s8-services.md>
+- **The open-items file** (entry "Left open by S8": the logs
+  follow-up and what unblocks it):
+  <https://github.com/prisma/prisma-cli/blob/main/.drive/projects/prisma-cli-v8/deferred.md>
+- **The consumer template the CLI will copy**:
+  <https://github.com/prisma/prisma-cli/blob/main/packages/cli/src/v8/build/logs.ts>
+  (`ctx.api.GET(..., parseAs: "stream")`, line-parsed records).
+
+The shelved CLI handler (reviewed and green before shelving) is in
+prisma-cli's `s2c-services` branch history; the CLI team restores and
+reshapes it against your pinned contract, with fixture-driven tests —
+so the CLI side can land before your deploy and light up when the
+endpoint ships.
diff --git a/.drive/projects/prisma-cli-v8/assets/s2/parity-divergences-service-logs.md b/.drive/projects/prisma-cli-v8/assets/s2/parity-divergences-service-logs.md
new file mode 100644
index 00000000..acad90fe
--- /dev/null
+++ b/.drive/projects/prisma-cli-v8/assets/s2/parity-divergences-service-logs.md
@@ -0,0 +1,109 @@
+# service logs parity divergences
+
+What changes for a user between the `service logs` that S2c shelved and
+the one this slice ships. Same entry format as
+[`parity-divergences.md`](parity-divergences.md), and the same standing
+ruling behind it (S2 standing ruling 10: divergences are enumerated, not
+discovered).
+
+**The baseline is the shelved S2c command**, not a shipping binary. No
+released CLI in this repo has had a working `service logs`: `app logs`
+died with the commander shell, and S2c's replacement never landed
+because the transport it needed did not exist. So nothing below breaks a
+user's existing habit — but the command's shape moved a long way from
+the one the S2c record describes, and that record is what a reader will
+have in hand.
+
+The command mounts as **`service logs`**, the legacy spelling, not under
+the `deployment` subgroup the S8 reshape created (ruled, operator,
+2026-08-13; recorded in `deferred.md`).
+
+## Reading a page replaces holding a socket
+
+S2c streamed: it opened a WebSocket through the compute SDK's
+`streamLogs`, authenticated it with a raw token it fetched itself, and
+printed records until the far end closed. There was no flag to ask for
+anything else, so **following was the only behaviour**.
+
+The platform serves one page per plain `GET` and closes it
+(pdp-control-plane #4886), so the command is a page read:
+
+- **Default: the last 100 lines, then exit 0.** The shape `kubectl logs`
+  has. A script that ran the S2c command and expected it to block until
+  interrupted now gets output and a clean exit instead.
+- **`--follow` asks for the streaming behaviour back**, and gets polling
+  rather than push: each page closes with a terminal record naming the
+  cursor the next page starts at, so the command waits two seconds and
+  asks again from there. The lines are the same; the latency is now
+  bounded by the poll interval rather than by the server's write.
+- The WebSocket upgrade still exists on the same path and the CLI never
+  uses it. The engine socket transport S2c waited for was not built
+  (R-S8-5), and the shelved design stays shelved for the live-streaming
+  date.
+
+## Three flags that did not exist
+
+`--tail <n>`, `--from-start` and the page-size default are all new
+surface — S2c had no way to say how much log to read.
+
+- `--tail <n>` resizes the page; unflagged runs send `tail=100`
+  explicitly, matching the endpoint's own default rather than relying
+  on it.
+- `--from-start` reads from the beginning instead.
+- **Passing both is refused** (`SERVICE.LOGS_RANGE_CONFLICT`, exit 2),
+  before the target is resolved or anything is read: they name opposite
+  ends of the same log, so there is no reading that satisfies both.
+
+`--deployment`, `--service`, `--project` and the config-target
+positional carry over from S2c unchanged, resolution and error shapes
+included.
+
+## Two refusals where S2c would have carried on
+
+Both are cases the platform contract says should not arise. They are
+refused rather than absorbed, because absorbing them produces output a
+user cannot tell from correct output.
+
+- **A page that closes with no resume cursor stops `--follow`**
+  (`SERVICE.LOGS_NO_CURSOR`). Continuing would mean re-requesting with no
+  range, which the endpoint answers with its default tail — the same
+  hundred lines reprinted every two seconds, indefinitely, with nothing
+  saying why.
+- **A page whose body ends without its terminal record is an incomplete
+  read** (`SERVICE.LOGS_INCOMPLETE`), in page mode as well as follow.
+  The lines that did arrive are still printed; what is refused is
+  settling as though the whole page had been read, which would show a
+  user a truncated log with no sign it was truncated.
+
+Neither settles quietly, and the reasoning for `--follow` is that it has
+no successful ending: it runs until interrupted, which settles 130, or
+it fails. An exit 0 from a follow would be a new outcome meaning "gave
+up", indistinguishable from a follow that never started.
+
+## An error terminal record ends the run
+
+`type: "terminal"` with `kind: "error"` is the platform reporting that
+the log read itself failed. It settles as `SERVICE.LOGS_FAILED` carrying
+the record's own code, message and `retryable` flag, exit 2 — where S2c
+printed the message to the diagnostic channel and exited 0.
+
+In `--follow`, a **retryable** error terminal is retried once after the
+poll interval, and the budget resets on any page that succeeds. So a
+long follow survives repeated transient failures but never loops on a
+persistent one.
+
+## Not a divergence, recorded because it looks like one
+
+**Interrupting `--follow` settles 130.** Ctrl-C ends the run at 128 +
+SIGINT from the engine's own record of the signal, so a wrapper script
+that treats a non-zero exit as failure sees one when a developer stops
+following.
+
+This is worth stating because the S2c handler reads as though it did the
+opposite — it treats a cancelled stream as an expected user action and
+returns success. That difference is not observable: the engine settles a
+signalled run at 128 + the signal "whatever the handler concluded"
+(operator ruling, 2026-08-11, recorded against `composer dev` in
+[`parity-divergences-s3.md`](parity-divergences-s3.md)). S2c would have
+settled 130 too. The exit code comes from the engine rule, not from
+anything this slice changed.
diff --git a/.drive/projects/prisma-cli-v8/deferred.md b/.drive/projects/prisma-cli-v8/deferred.md
index fed5bb82..cb355fee 100644
--- a/.drive/projects/prisma-cli-v8/deferred.md
+++ b/.drive/projects/prisma-cli-v8/deferred.md
@@ -127,6 +127,12 @@ CLI does not do, and each restarts as engine work if wanted:
   with the commander shell and `service logs` waits on an engine
   streaming transport. The one capability loss of S2d; the S2c record
   has the design notes.
+  **Superseded by the `service-logs` slice (2026-08-13):** `service logs`
+  ships, reading the platform's HTTP page endpoint, so the capability is
+  no longer missing. What is still missing is the *streaming* half —
+  `--follow` polls on a 2s interval rather than holding a socket open.
+  The open remainder is the WebSocket live tail, in the closed
+  `service logs` entry further down this file.
 - **`build logs` cannot exit 1 on a failed build** until the engine
   grows a way for a stream to settle with a documented non-zero code.
 - **The crash-recovery feedback action does not port** (the legacy
@@ -193,6 +199,16 @@ CLI does not do, and each restarts as engine work if wanted:
   `packages/cli`. Fix: have the child write `ready` only after the
   handler is installed — the same ordering the engine's own
   `tests/fixtures/child.mjs` `trap-term` fixture already uses.
+- **`credential-manager.test.ts`'s crashed-lock contention test is
+  flaky on Windows CI.** "lets only one of two waiting mutations
+  clear the same crashed holder's lock" timed out at 5 s on
+  `windows-latest` during #181 (2026-08-13, a PR touching nothing
+  near the credential manager) and passed on the re-run of the same
+  commit. One sighting so far — same reddens-shared-checks family as
+  the two entries above. Likely shape: two waiters racing a lock
+  file under Windows FS latency needs more than the 5 s budget, or
+  the same write-marker-before-ready ordering. Diagnose on the
+  second sighting.
 
 ## Owned by whoever converts a family's renderers
 
@@ -305,18 +321,26 @@ CLI does not do, and each restarts as engine work if wanted:
   `managedBy` marker — the same request deferred alongside it. Users own
   their resources, and Composer reconciling a manual change on the next
   deploy is accepted behavior until then.
-- **`service logs` is a follow-up slice, and its transport question is
-  ANSWERED (R-S8-5).** API owners, via operator, 2026-08-12: **HTTP
-  instead of WebSocket is acceptable, provided live streaming can be
-  added at a later date.** So no engine socket transport was built. The
-  command lands as a copy of `build logs` — plain HTTP,
-  `parseAs: "stream"` — in a follow-up slice, once the platform endpoint
-  serves HTTP. The engine WebSocket design
-  (`assets/engine/websocket-transport-design.md`) is **shelved as the
-  later live-streaming path, not deleted.** The ownership question the
-  plan raised dissolved on investigation: `composer log` attaches to the
-  local dev daemon's streams, a `service deployment logs` would read the
-  platform endpoint — different data, no shared subgroup.
+- **`service logs` SHIPPED** (slice `service-logs`, 2026-08-13 —
+  contract at `specs/service-logs.md`, divergences at
+  `assets/s2/parity-divergences-service-logs.md`). It mounts as
+  `service logs`, the legacy spelling (ruled, operator, 2026-08-13), and
+  reads the platform's HTTP page endpoint (pdp-control-plane #4886): one
+  page by default, `--follow` polling on the terminal record's cursor.
+  Closing this entry corrects one thing it predicted: the pinned
+  `@prisma/management-api-sdk` (1.55.0) did NOT need a bump. The
+  `query: never` the risk note named is path-item boilerplate that every
+  path in that file carries; the operation type
+  (`getV1DeploymentsByDeploymentIdLogs`) already declared `tail`,
+  `from_start` and `cursor`, so the wiring typechecked against the
+  existing pin with no cast.
+  **What stays open: the WebSocket live tail.** The platform serves the
+  upgrade on the same path and the CLI does not use it, so following is
+  polling on a 2s interval rather than push. The engine socket design
+  (`assets/engine/websocket-transport-design.md`) remains **shelved for
+  that later date, not deleted** — R-S8-5's "provided live streaming can
+  be added at a later date" is still the standing commitment, and this
+  slice is what it was traded against.
 - **The e2e suite should assert the real service-id prefix.** D2 wrote
   `e2e/service.e2e.ts` without credentials to run it, so it asserts only
   that `service create` reports a non-empty id. The sibling suites assert
diff --git a/.drive/projects/prisma-cli-v8/plans/service-logs.md b/.drive/projects/prisma-cli-v8/plans/service-logs.md
new file mode 100644
index 00000000..04d0a1e3
--- /dev/null
+++ b/.drive/projects/prisma-cli-v8/plans/service-logs.md
@@ -0,0 +1,26 @@
+# service logs dispatch plan
+
+Contract: `../specs/service-logs.md`. Branch `service-logs` off
+`main`. One PR.
+
+## D1 — the command
+
+Outcome: `service logs` mounted and green per the contract — the
+S2c resolution logic restored from `bot/s2c-services`, the transport
+replaced with the page-read GET, page and follow modes, the full
+test matrix on fixtures and the injectable clock. STOP if the pinned
+SDK's `query: never` blocks typecheck without a cast.
+
+Builds on: main. Hands to D2: the command green, any SDK blocker
+named.
+
+Completed when: contract acceptance items 1–3; gate green
+(engine suite, cli suite, typecheck, lint — sequential, exit 0).
+
+## D2 — closure
+
+Outcome: divergence entry, `deferred.md`'s logs entry closes, e2e
+backlog entry, review loop, PR. Reconcile with records PR #176's
+edit of the same `deferred.md` entry if it has merged.
+
+Completed when: acceptance items 4–6; gate green.
diff --git a/.drive/projects/prisma-cli-v8/specs/service-logs.md b/.drive/projects/prisma-cli-v8/specs/service-logs.md
new file mode 100644
index 00000000..826d37e4
--- /dev/null
+++ b/.drive/projects/prisma-cli-v8/specs/service-logs.md
@@ -0,0 +1,116 @@
+# service logs (slice contract)
+
+Status: rev 1 (2026-08-13). One PR into `main`, branch
+`service-logs`. Repo: prisma-cli only. Unshelves the S2c `service
+logs` command against the platform's new HTTP page-read contract
+(pdp-control-plane PR #4886, the base for this slice).
+
+Operator rulings carried in: the command mounts as **`service logs`**
+(legacy spelling — ruled 2026-08-13, recorded in `deferred.md`); no
+engine WebSocket transport (R-S8-5; the WS live tail stays the
+platform's, unused by the CLI until the live-streaming date).
+
+## The endpoint contract (PR #4886, pinned)
+
+`GET /v1/deployments/{deploymentId}/logs`, authenticated, plain GET:
+one request returns ONE PAGE as `application/x-ndjson` — records
+`{type:"log", text, byteStart, byteEnd}` — and ends with
+`{type:"terminal", kind:"end"|"error", code, message, retryable,
+cursor}` before closing. Query: `tail=N` (last N lines, default
+100), `from_start=true` (page from the beginning), `cursor`
+(continue a chain; the terminal record's cursor is the next start).
+No held-open connection; the WebSocket upgrade on the same path is
+out of scope for the CLI.
+
+## The command
+
+`service logs [<service>] [--service name] [--project id-or-name]
+[--deployment id] [--tail n] [--from-start] [--follow]`
+
+- Session command in the platform family's service group. Target
+  resolution ports VERBATIM from the shelved S2c handler
+  (`bot/s2c-services`, `packages/cli/src/v8/service/logs.ts`):
+  explicit `--deployment` resolved globally then checked against the
+  project; otherwise the service's live deployment; the S2c error
+  shapes (`deploymentNotFoundError`, `deploymentOutsideProjectError`,
+  `noDeploymentsError`, …) return with it. The dead parts do NOT
+  port: `streamLogs`/compute-sdk, `getApiBaseUrl`,
+  `logStreamCredentialsError` (no credential ever reaches the
+  command — the transport is `ctx.api`).
+- **Default: one page, then exit 0** — `tail` 100 like the endpoint,
+  the kubectl-logs shape. `--tail n` passes through; `--from-start`
+  maps to `from_start=true` (constructing it with `--tail` is a
+  parse-time conflict). A routine terminal record (`kind:"end"`)
+  ends the page; its cursor is not surfaced (the CLI owns resume).
+- **`--follow`**: after each `kind:"end"` terminal record, wait the
+  poll interval and re-request with that record's cursor; run until
+  the user interrupts (the engine settles 130 from its signal
+  record, as `dev` does). Poll interval 2 s on the injectable clock.
+- Records map per `build logs` (R-S2c-2): `type:"log"` → `output`
+  events, channel `data`; json mode frames them (session kind).
+  `type:"terminal"` with `kind:"error"` → structured error carrying
+  the record's code/message, exit non-zero; `retryable: true` on an
+  error terminal in `--follow` mode retries ONCE after the interval,
+  then fails (do not loop on a persistent error).
+- Transport: `ctx.api.GET("/v1/deployments/{deploymentId}/logs",
+  { parseAs: "stream", params: { query: ... } })` — the `build logs`
+  shape, line-split NDJSON, tolerant of a final partial line.
+
+## The SDK risk — RETIRED at D1 (the premise was wrong)
+
+Rev 1 claimed the pinned SDK types this path's `query` as `never`.
+That read the path-item boilerplate (identical on every path); the
+OPERATION type (`getV1DeploymentsByDeploymentIdLogs`,
+`dist/index.d.ts:6188` in `@prisma/management-api-sdk@1.55.0`)
+already publishes `tail?: number`, `from_start?: "true" | "false"`
+(a string union — the command sends `"true"`), and
+`cursor?: string`. Verified empirically at D1 under `pnpm
+typecheck`. The stream body keeps `build logs`' established cast
+(the spec documents no 200 body); no new cast kind.
+
+## D1 amendments (implementer decisions, orchestrator-ratified)
+
+- The `--tail`/`--from-start` conflict refuses at HANDLER TOP per
+  the `project transfer` precedent (the engine has no declarative
+  flag-conflict mechanism), as `SERVICE.LOGS_RANGE_CONFLICT`, exit
+  2, before any request.
+- The poll interval is `PRISMA_CLI_SERVICE_LOGS_POLL_MS` per the
+  `service domain wait` precedent — the engine's delay seam is not
+  reachable from `CommandContext`; exposing it is an engine change
+  this slice does not make.
+- `build logs`' private NDJSON reader moved to `lib/ndjson.ts`,
+  shared by both commands — an extraction, not a behavior change;
+  duplicating chunk-boundary handling is the drift class the S8
+  workspace-filter defect came from.
+- Follow-mode retry: a retryable error terminal is retried once per
+  FAILURE, with the budget reset by any successful page — a long
+  follow survives repeated transients but never loops on a
+  persistent error.
+- e2e: `EXCLUSIONS` (needs a Composer-deployed service), matching
+  the S8 lifecycle commands — supersedes acceptance item 6's
+  "backlog" wording.
+
+## Out of scope
+
+The WebSocket live tail (later date, platform's move); any engine
+transport work; `composer log` (different data source — the local
+dev daemon); changes to `build logs`.
+
+## Acceptance
+
+- [ ] `service logs` mounted (legacy spelling), group help updated;
+      grammar per above with the `--tail`/`--from-start` conflict at
+      parse time.
+- [ ] Page mode: fixture-backed tests for tail default, `--tail`,
+      `--from-start`, explicit `--deployment`, the S2c resolution
+      errors, unframed data output, json framing, and the error
+      terminal record → structured error.
+- [ ] Follow mode: fixture drives page → end(cursor) → page →
+      interrupt on the injectable clock; cursor passed correctly;
+      retryable-error single retry pinned; interrupt settles 130.
+- [ ] Divergence entry (`assets/s2/parity-divergences-s8.md` gains a
+      follow-up section or a new sibling file): default is page-read
+      (legacy followed); `--follow` is polling, not push.
+- [ ] `deferred.md`'s logs entry closes; e2e joins the
+      deployed-service backlog beside `service open`.
+- [ ] Suites green sequentially; typecheck + lint exit 0.
diff --git a/packages/cli/src/cli.ts b/packages/cli/src/cli.ts
index dfb77eb6..1af779c1 100644
--- a/packages/cli/src/cli.ts
+++ b/packages/cli/src/cli.ts
@@ -77,6 +77,7 @@ import { serviceDomainRetryCommand } from "./commands/service/domain-retry";
 import { serviceDomainShowCommand } from "./commands/service/domain-show";
 import { serviceDomainWaitCommand } from "./commands/service/domain-wait";
 import { serviceListCommand } from "./commands/service/list";
+import { serviceLogsCommand } from "./commands/service/logs";
 import { serviceOpenCommand } from "./commands/service/open";
 import { serviceRemoveCommand } from "./commands/service/remove";
 import { serviceShowCommand } from "./commands/service/show";
@@ -122,6 +123,7 @@ export const platformCommandFamily: CommandFamily = defineCommandFamily({
     gitConnect: gitConnectCommand,
     gitDisconnect: gitDisconnectCommand,
     serviceList: serviceListCommand,
+    serviceLogs: serviceLogsCommand,
     serviceCreate: serviceCreateCommand,
     serviceShow: serviceShowCommand,
     serviceOpen: serviceOpenCommand,
@@ -238,6 +240,7 @@ export const mountedCommands: Readonly<Record<string, AnyCommand>> = {
   "git connect": gitConnectCommand,
   "git disconnect": gitDisconnectCommand,
   "service list": serviceListCommand,
+  "service logs": serviceLogsCommand,
   "service create": serviceCreateCommand,
   "service show": serviceShowCommand,
   "service open": serviceOpenCommand,
diff --git a/packages/cli/src/commands/build/logs.ts b/packages/cli/src/commands/build/logs.ts
index c18e69b7..99606e7f 100644
--- a/packages/cli/src/commands/build/logs.ts
+++ b/packages/cli/src/commands/build/logs.ts
@@ -2,6 +2,7 @@ import type { CommandContext } from "@prisma/cli-engine";
 import { defineSessionCommand, flag, positional } from "@prisma/cli-engine";
 import { CliStructuredError, ok } from "@prisma/cli-engine/protocol";
 import { CLI_NAME } from "../../cli-name";
+import { forEachNdjsonRecord } from "../../lib/ndjson";
 
 const TRAILING_NEWLINE = /\n$/;
 
@@ -105,51 +106,6 @@ function buildFailedError(
   });
 }
 
-/** Reads a newline-delimited JSON body line by line. */
-async function forEachNdjsonRecord<T>(
-  body: ReadableStream<Uint8Array>,
-  onRecord: (record: T) => void,
-): Promise<void> {
-  const reader = body.getReader();
-  const decoder = new TextDecoder();
-  let buffer = "";
-
-  // An abort, a malformed line, or a throwing onRecord all leave the
-  // loop without reaching `done`. Cancelling closes the HTTP body
-  // instead of holding the socket open until garbage collection, which
-  // a long `--follow` run makes observable.
-  try {
-    for (;;) {
-      // biome-ignore lint/performance/noAwaitInLoops: a stream must be read sequentially, chunk by chunk.
-      const { done, value } = await reader.read();
-      if (value) {
-        buffer += decoder.decode(value, { stream: true });
-      }
-
-      let newlineIndex = buffer.indexOf("\n");
-      while (newlineIndex !== -1) {
-        const line = buffer.slice(0, newlineIndex).trim();
-        buffer = buffer.slice(newlineIndex + 1);
-        if (line) {
-          onRecord(JSON.parse(line) as T);
-        }
-        newlineIndex = buffer.indexOf("\n");
-      }
-
-      if (done) {
-        const tail = buffer.trim();
-        if (tail) {
-          onRecord(JSON.parse(tail) as T);
-        }
-        return;
-      }
-    }
-  } finally {
-    await reader.cancel().catch(() => undefined);
-    reader.releaseLock();
-  }
-}
-
 function reportRecord(
   ctx: Pick<CommandContext, "report">,
   record: BuildLogRecord,
diff --git a/packages/cli/src/commands/service/errors.ts b/packages/cli/src/commands/service/errors.ts
index eaa66dbd..3fa92799 100644
--- a/packages/cli/src/commands/service/errors.ts
+++ b/packages/cli/src/commands/service/errors.ts
@@ -200,6 +200,61 @@ export function deploymentNotFoundError(
   );
 }
 
+/** `--tail` and `--from-start` ask for opposite ends of the log, so a
+ *  run naming both has no answer to give. Refused before any work. */
+export function logsRangeConflictError(): CliStructuredError {
+  return new CliStructuredError(
+    "SERVICE.LOGS_RANGE_CONFLICT",
+    "Choose one end of the log to read from",
+    {
+      why: "--tail and --from-start are mutually exclusive: one reads the last lines, the other reads from the beginning.",
+      nextActions: [
+        adviceAction(
+          "Pass --tail <n> for the last n lines, or --from-start for the whole log.",
+        ),
+      ],
+    },
+  );
+}
+
+/** A deployment id resolves globally, so one that exists but belongs to
+ *  another project is its own failure — not "not found". */
+export function deploymentOutsideProjectError(
+  deploymentId: string,
+): CliStructuredError {
+  return new CliStructuredError(
+    "SERVICE.DEPLOYMENT_OUTSIDE_PROJECT",
+    `Deployment "${deploymentId}" belongs to another project`,
+    {
+      why: "The deployment exists, but the service that owns it is not in the resolved project.",
+      nextActions: [
+        adviceAction("Pass --project for the project that owns it."),
+        runCommandAction("List services", "service list"),
+      ],
+    },
+  );
+}
+
+/** The deployment exists but names no owning service, so there is no
+ *  project to check it against and nothing to scope logs by. */
+export function deploymentDetachedError(
+  deploymentId: string,
+): CliStructuredError {
+  return new CliStructuredError(
+    "SERVICE.DEPLOYMENT_DETACHED",
+    `Deployment "${deploymentId}" has no owning service`,
+    {
+      why: "The Management API returned the deployment without a service, so it cannot be scoped to a project.",
+      nextActions: [
+        runCommandAction(
+          "Show the deployment",
+          `service deployment show ${deploymentId}`,
+        ),
+      ],
+    },
+  );
+}
+
 export function deploymentNotFoundForServiceError(
   deploymentId: string,
   serviceName: string,
diff --git a/packages/cli/src/commands/service/logs.ts b/packages/cli/src/commands/service/logs.ts
new file mode 100644
index 00000000..e20c1c55
--- /dev/null
+++ b/packages/cli/src/commands/service/logs.ts
@@ -0,0 +1,504 @@
+import type { CommandContext } from "@prisma/cli-engine";
+import { defineSessionCommand, flag, positional } from "@prisma/cli-engine";
+import { CliStructuredError, ok } from "@prisma/cli-engine/protocol";
+import type { AppProvider, AppRecord } from "../../lib/app/app-provider";
+import { forEachNdjsonRecord } from "../../lib/ndjson";
+import {
+  adviceAction,
+  deployFailedError,
+  deploymentDetachedError,
+  deploymentNotFoundError,
+  deploymentOutsideProjectError,
+  logsRangeConflictError,
+  noDeploymentsError,
+  runCommandAction,
+} from "./errors";
+import { requireDeploymentForService } from "./release";
+import type { ServiceDeploymentSummary } from "./results";
+import type { ServiceContext, ServiceReadState } from "./target";
+import {
+  applyLiveDeploymentHint,
+  listServices,
+  rememberSelectedService,
+  resolveCurrentLiveDeploymentId,
+  resolveServiceReadState,
+} from "./target";
+
+const TRAILING_NEWLINE = /\n$/;
+
+/** The endpoint's own default page size, restated so `--tail` and the
+ *  unflagged run send the same shape of request. */
+const DEFAULT_TAIL = 100;
+
+/** Contract: poll every 2s in --follow. Overridable so a test drives the
+ *  loop without waiting, the way `service domain wait` does. */
+const DEFAULT_POLL_INTERVAL_MS = 2000;
+
+function pollIntervalMs(ctx: ServiceContext): number {
+  const raw = ctx.env.PRISMA_CLI_SERVICE_LOGS_POLL_MS;
+  if (!raw) {
+    return DEFAULT_POLL_INTERVAL_MS;
+  }
+  const parsed = Number.parseInt(raw, 10);
+  return Number.isInteger(parsed) && parsed >= 0
+    ? parsed
+    : DEFAULT_POLL_INTERVAL_MS;
+}
+
+async function sleep(milliseconds: number, signal: AbortSignal): Promise<void> {
+  if (milliseconds <= 0) {
+    signal.throwIfAborted();
+    return;
+  }
+  signal.throwIfAborted();
+  await new Promise<void>((resolve, reject) => {
+    const onAbort = () => {
+      clearTimeout(timeout);
+      reject(signal.reason);
+    };
+    const timeout = setTimeout(() => {
+      signal.removeEventListener("abort", onAbort);
+      resolve();
+    }, milliseconds);
+    signal.addEventListener("abort", onAbort, { once: true });
+  });
+}
+
+/**
+ * One line of `GET /v1/deployments/{deploymentId}/logs`. A page is a run
+ * of `log` records closed by exactly one `terminal`, whose cursor is
+ * where the next page starts.
+ */
+type DeploymentLogRecord =
+  | { type: "log"; text: string; byteStart: number; byteEnd: number }
+  | {
+      type: "terminal";
+      kind: "end" | "error";
+      code: string;
+      message: string;
+      retryable: boolean;
+      cursor: string | null;
+    };
+
+type TerminalRecord = Extract<DeploymentLogRecord, { type: "terminal" }>;
+
+interface LogTarget {
+  service: AppRecord;
+  deployment: ServiceDeploymentSummary;
+}
+
+function logsFailedError(
+  deploymentId: string,
+  status: number,
+): CliStructuredError {
+  return new CliStructuredError(
+    "SERVICE.LOGS_FAILED",
+    `Failed to read logs for deployment ${deploymentId}`,
+    {
+      why: `The Management API returned HTTP ${status}.`,
+      meta: { status },
+      nextActions: [
+        adviceAction(
+          "Retry the command, or rerun with --log-level verbose for more detail.",
+        ),
+        runCommandAction(
+          "Show the deployment",
+          `service deployment show ${deploymentId}`,
+        ),
+      ],
+    },
+  );
+}
+
+/**
+ * The body ended mid-page, without the terminal record that closes one.
+ * Distinct from SERVICE.LOGS_NO_CURSOR, which is a page that closed
+ * properly and said there is nothing to resume from: this one is an
+ * incomplete read, and the lines already printed are not the whole page.
+ */
+function logsIncompleteError(deploymentId: string): CliStructuredError {
+  return new CliStructuredError(
+    "SERVICE.LOGS_INCOMPLETE",
+    `Incomplete log page for deployment ${deploymentId}`,
+    {
+      why: "The response ended without the record that closes a page, so the lines shown may be only part of it.",
+      nextActions: [adviceAction("Rerun the command to read the page again.")],
+    },
+  );
+}
+
+/** An error terminal record is the platform reporting that the log read
+ *  itself failed, so it settles the run rather than printing. */
+function logStreamFailedError(
+  deploymentId: string,
+  record: TerminalRecord,
+): CliStructuredError {
+  return new CliStructuredError(
+    "SERVICE.LOGS_FAILED",
+    `Log stream failed for deployment ${deploymentId}`,
+    {
+      why: record.message,
+      meta: {
+        code: record.code,
+        retryable: record.retryable,
+        ...(record.cursor === null ? {} : { cursor: record.cursor }),
+      },
+      nextActions: [
+        runCommandAction(
+          "Show the deployment",
+          `service deployment show ${deploymentId}`,
+        ),
+      ],
+    },
+  );
+}
+
+function listDeployments(
+  ctx: ServiceContext,
+  provider: AppProvider,
+  serviceId: string,
+) {
+  return provider
+    .listDeployments(serviceId, { signal: ctx.signal })
+    .catch((error): never => {
+      throw deployFailedError("Failed to list service deployments", error, [
+        runCommandAction("List deployments", "service deployment list"),
+      ]);
+    });
+}
+
+/** `--deployment <id>`: the id is global, so it is resolved directly and
+ *  then checked against the resolved project — a deployment that exists
+ *  but belongs elsewhere is reported as its own failure. */
+async function resolveExplicitDeployment(
+  ctx: ServiceContext,
+  state: ServiceReadState,
+  serviceName: string | undefined,
+  deploymentId: string,
+): Promise<LogTarget> {
+  if (serviceName) {
+    if (!state.selected) {
+      throw noDeploymentsError(
+        "No deployments available to read logs from",
+        "The resolved project does not have any deployed service yet.",
+      );
+    }
+    const deploymentsResult = await listDeployments(
+      ctx,
+      state.provider,
+      state.selected.id,
+    );
+    const deployment = requireDeploymentForService(
+      deploymentsResult.deployments,
+      deploymentId,
+      state.selected.name,
+    );
+    await rememberSelectedService(
+      state.stateStore,
+      state.projectId,
+      deploymentsResult.app,
+    );
+    return { service: deploymentsResult.app, deployment };
+  }
+
+  const shown = await state.provider
+    .showDeployment(deploymentId, { signal: ctx.signal })
+    .catch((error) => {
+      throw deployFailedError("Failed to show deployment", error, [
+        runCommandAction("List deployments", "service deployment list"),
+      ]);
+    });
+  if (!shown) {
+    throw deploymentNotFoundError(deploymentId);
+  }
+  if (!shown.app) {
+    throw deploymentDetachedError(deploymentId);
+  }
+
+  const services = await listServices(
+    ctx,
+    state.provider,
+    state.projectId,
+    state.target.branch.name,
+  );
+  const owning = services.find((service) => service.id === shown.app?.id);
+  if (!owning) {
+    throw deploymentOutsideProjectError(deploymentId);
+  }
+
+  await rememberSelectedService(state.stateStore, state.projectId, owning);
+  return { service: owning, deployment: shown.deployment };
+}
+
+/** No `--deployment`: read whatever is live for the selected service. */
+async function resolveLiveDeployment(
+  ctx: ServiceContext,
+  state: ServiceReadState,
+): Promise<LogTarget> {
+  if (!state.selected) {
+    throw noDeploymentsError(
+      "No deployments available to read logs from",
+      "The resolved project does not have any deployed service yet.",
+    );
+  }
+
+  const deploymentsResult = await listDeployments(
+    ctx,
+    state.provider,
+    state.selected.id,
+  );
+  const currentLiveDeploymentId = resolveCurrentLiveDeploymentId(
+    deploymentsResult.app,
+    deploymentsResult.deployments,
+  );
+  const deployments = applyLiveDeploymentHint(
+    deploymentsResult.deployments,
+    currentLiveDeploymentId,
+  );
+  const deployment = currentLiveDeploymentId
+    ? (deployments.find(
+        (candidate) => candidate.id === currentLiveDeploymentId,
+      ) ?? null)
+    : null;
+
+  await rememberSelectedService(
+    state.stateStore,
+    state.projectId,
+    deploymentsResult.app,
+  );
+
+  if (!deployment) {
+    throw noDeploymentsError(
+      "No deployments available to read logs from",
+      `The selected service "${deploymentsResult.app.name}" does not have a live deployment.`,
+    );
+  }
+  return { service: deploymentsResult.app, deployment };
+}
+
+/**
+ * Reads one page and reports its log records. Returns the terminal
+ * record that closed it — the caller decides whether that ends the run
+ * or starts the next page.
+ *
+ * Every page ends with a terminal record, so a body that stops without
+ * one was truncated. The lines that did arrive have already been
+ * reported, but the run must not settle as though it had read the whole
+ * page: the user would have a partial log and no way to tell.
+ */
+async function readPage(
+  ctx: CommandContext,
+  deploymentId: string,
+  query: { tail?: number; from_start?: "true"; cursor?: string },
+): Promise<TerminalRecord> {
+  const { data, response } = await ctx.api.GET(
+    "/v1/deployments/{deploymentId}/logs",
+    {
+      params: { path: { deploymentId }, query },
+      parseAs: "stream",
+      signal: ctx.signal,
+    },
+  );
+
+  const body = data as ReadableStream<Uint8Array> | null | undefined;
+  if (!response.ok || !body) {
+    await body?.cancel().catch(() => undefined);
+    throw response.status === 404
+      ? deploymentNotFoundError(deploymentId)
+      : logsFailedError(deploymentId, response.status);
+  }
+
+  let terminal: TerminalRecord | null = null;
+  await forEachNdjsonRecord<DeploymentLogRecord>(body, (record) => {
+    if (record.type === "terminal") {
+      terminal = record;
+      return;
+    }
+    ctx.report({
+      kind: "output",
+      source: "logs",
+      channel: "data",
+      line: record.text.replace(TRAILING_NEWLINE, ""),
+      data: { byteStart: record.byteStart, byteEnd: record.byteEnd },
+    });
+  });
+  if (terminal === null) {
+    throw logsIncompleteError(deploymentId);
+  }
+  return terminal;
+}
+
+/**
+ * Following needs somewhere to resume from. Without a cursor the next
+ * request would carry no range at all, the endpoint would apply its
+ * default tail, and the same lines would print again every interval —
+ * silent duplication the user cannot act on. So the run stops and says
+ * why. It settles as an error rather than a clean end because `--follow`
+ * has no successful ending: it runs until interrupted (130) or fails,
+ * and an exit 0 here would be a novel outcome meaning "gave up".
+ */
+function requireResumeCursor(
+  deploymentId: string,
+  cursor: string | null,
+): string {
+  if (cursor === null) {
+    throw new CliStructuredError(
+      "SERVICE.LOGS_NO_CURSOR",
+      `Cannot follow logs for deployment ${deploymentId}`,
+      {
+        why: "The log page ended without a resume cursor, so there is no point to continue reading from.",
+        nextActions: [
+          adviceAction(
+            "Rerun without --follow to read the page, or retry if the deployment is still starting.",
+          ),
+        ],
+      },
+    );
+  }
+  return cursor;
+}
+
+/**
+ * `--follow`: wait the poll interval, read the next page from the cursor
+ * the last one ended on, repeat until the user interrupts. Never
+ * returns — the run ends by abort (the engine settles 130) or by throw.
+ */
+async function followPages(
+  ctx: CommandContext,
+  deploymentId: string,
+  startCursor: string | null,
+): Promise<never> {
+  const interval = pollIntervalMs(ctx);
+  let cursor = requireResumeCursor(deploymentId, startCursor);
+  // One retry, not a loop: a retryable error that keeps happening is a
+  // persistent failure, and hammering it would hide that. Any page that
+  // succeeds restores the budget, so a long follow survives repeated
+  // transients without ever looping on a persistent one.
+  let retriedAfterError = false;
+
+  for (;;) {
+    // biome-ignore lint/performance/noAwaitInLoops: --follow polls one page at a time; the wait between reads is the point, and each page starts at the cursor the previous one ended on.
+    await sleep(interval, ctx.signal);
+    const next = await readPage(ctx, deploymentId, { cursor });
+
+    if (next.kind === "error") {
+      if (!next.retryable || retriedAfterError) {
+        throw logStreamFailedError(deploymentId, next);
+      }
+      retriedAfterError = true;
+      continue;
+    }
+    retriedAfterError = false;
+    cursor = requireResumeCursor(deploymentId, next.cursor);
+  }
+}
+
+export const serviceLogsCommand = defineSessionCommand({
+  help: {
+    summary: "Read logs for a deployment of the service",
+    examples: [
+      "service logs",
+      "service logs --tail 500",
+      "service logs --follow",
+      "service logs --deployment dep_123 --from-start",
+    ],
+  },
+  args: {
+    flags: {
+      service: flag.string({ brief: "Service name", placeholder: "name" }),
+      project: flag.string({
+        brief: "Project id or name",
+        placeholder: "id-or-name",
+      }),
+      deployment: flag.string({
+        brief: "Deployment id to read (default: the live deployment)",
+        placeholder: "id",
+      }),
+      tail: flag.number({
+        brief: `Read the last N lines (default ${DEFAULT_TAIL})`,
+        placeholder: "n",
+      }),
+      fromStart: flag.boolean({
+        brief: "Read from the beginning instead of the last lines",
+      }),
+      follow: flag.boolean({
+        brief: "Keep polling for new lines until interrupted",
+      }),
+    },
+    positionals: {
+      service: positional.optionalString({
+        brief:
+          "Service target from prisma.compute.ts when the config defines multiple services",
+        placeholder: "service",
+      }),
+    },
+  },
+  needs: { credentials: true },
+  handler: async (args, ctx) => {
+    // Refused before any work: the two ask for opposite ends of the log,
+    // so a run naming both has no answer to give.
+    if (args.flags.fromStart && args.flags.tail !== undefined) {
+      throw logsRangeConflictError();
+    }
+
+    // "A service was named" — by --service or by the config target. It
+    // decides whether an explicit deployment id is looked up within that
+    // service or resolved globally, and a global lookup needs no service
+    // selection at all (so it never prompts for one).
+    const serviceNamed = args.flags.service ?? args.positionals.service;
+    const resolveGlobally = Boolean(args.flags.deployment) && !serviceNamed;
+    const state = await resolveServiceReadState(ctx, {
+      ...(args.flags.service !== undefined
+        ? { serviceName: args.flags.service }
+        : {}),
+      ...(args.flags.project !== undefined
+        ? { projectRef: args.flags.project }
+        : {}),
+      ...(args.positionals.service !== undefined
+        ? { configTarget: args.positionals.service }
+        : {}),
+      commandName: "service logs",
+      skipSelection: resolveGlobally,
+    });
+
+    const target = args.flags.deployment
+      ? await resolveExplicitDeployment(
+          ctx,
+          state,
+          serviceNamed,
+          args.flags.deployment,
+        )
+      : await resolveLiveDeployment(ctx, state);
+    const deploymentId = target.deployment.id;
+
+    for (const line of [
+      `project: ${state.projectId}`,
+      `service: ${target.service.name}`,
+      `deployment: ${deploymentId}`,
+    ]) {
+      ctx.report({
+        kind: "output",
+        source: "logs",
+        channel: "diagnostic",
+        line,
+      });
+    }
+
+    const firstPageQuery: { tail?: number; from_start?: "true" } = args.flags
+      .fromStart
+      ? { from_start: "true" }
+      : { tail: args.flags.tail ?? DEFAULT_TAIL };
+
+    const terminal = await readPage(ctx, deploymentId, firstPageQuery);
+    if (terminal.kind === "error") {
+      throw logStreamFailedError(deploymentId, terminal);
+    }
+    if (!args.flags.follow) {
+      // The routine terminal record ends the page. Its cursor is the
+      // CLI's to resume from, not something the user is asked to carry.
+      return ok(undefined);
+    }
+
+    return followPages(ctx, deploymentId, terminal.cursor);
+  },
+});
diff --git a/packages/cli/src/commands/service/target.ts b/packages/cli/src/commands/service/target.ts
index d1630860..51a522aa 100644
--- a/packages/cli/src/commands/service/target.ts
+++ b/packages/cli/src/commands/service/target.ts
@@ -586,6 +586,10 @@ export async function resolveServiceReadState(
     configTarget?: string;
     branchName?: string;
     commandName: string;
+    /** Skip the service picker entirely. A caller resolving its target
+     *  by a globally-unique deployment id does not use the selection,
+     *  and selecting first would prompt for something it ignores. */
+    skipSelection?: boolean;
   },
 ): Promise<ServiceReadState> {
   const compute = await resolveComputeManagementContext(
@@ -609,13 +613,15 @@ export async function resolveServiceReadState(
     projectId,
     target.branch.name,
   );
-  const selected = await resolveExistingServiceSelection(
-    ctx,
-    stateStore,
-    projectId,
-    services,
-    options.serviceName ?? compute.configServiceName,
-  );
+  const selected = options.skipSelection
+    ? null
+    : await resolveExistingServiceSelection(
+        ctx,
+        stateStore,
+        projectId,
+        services,
+        options.serviceName ?? compute.configServiceName,
+      );
   return { provider, stateStore, target, projectId, selected };
 }
 
diff --git a/packages/cli/src/lib/ndjson.ts b/packages/cli/src/lib/ndjson.ts
new file mode 100644
index 00000000..2bfbcfad
--- /dev/null
+++ b/packages/cli/src/lib/ndjson.ts
@@ -0,0 +1,52 @@
+/**
+ * Reads a newline-delimited JSON body line by line.
+ *
+ * Shared by every command that reads an NDJSON log page, so the stream
+ * handling below is written and tested once. The two subtleties are the
+ * reason: a chunk boundary can fall inside a line, and a body can end
+ * without a trailing newline, so the last record arrives only if the
+ * leftover buffer is flushed at `done`.
+ */
+export async function forEachNdjsonRecord<T>(
+  body: ReadableStream<Uint8Array>,
+  onRecord: (record: T) => void,
+): Promise<void> {
+  const reader = body.getReader();
+  const decoder = new TextDecoder();
+  let buffer = "";
+
+  // An abort, a malformed line, or a throwing onRecord all leave the
+  // loop without reaching `done`. Cancelling closes the HTTP body
+  // instead of holding the socket open until garbage collection, which
+  // a long `--follow` run makes observable.
+  try {
+    for (;;) {
+      // biome-ignore lint/performance/noAwaitInLoops: a stream must be read sequentially, chunk by chunk.
+      const { done, value } = await reader.read();
+      if (value) {
+        buffer += decoder.decode(value, { stream: true });
+      }
+
+      let newlineIndex = buffer.indexOf("\n");
+      while (newlineIndex !== -1) {
+        const line = buffer.slice(0, newlineIndex).trim();
+        buffer = buffer.slice(newlineIndex + 1);
+        if (line) {
+          onRecord(JSON.parse(line) as T);
+        }
+        newlineIndex = buffer.indexOf("\n");
+      }
+
+      if (done) {
+        const tail = buffer.trim();
+        if (tail) {
+          onRecord(JSON.parse(tail) as T);
+        }
+        return;
+      }
+    }
+  } finally {
+    await reader.cancel().catch(() => undefined);
+    reader.releaseLock();
+  }
+}
diff --git a/packages/cli/tests/e2e-coverage.test.ts b/packages/cli/tests/e2e-coverage.test.ts
index 5cc48b87..eb5c62b8 100644
--- a/packages/cli/tests/e2e-coverage.test.ts
+++ b/packages/cli/tests/e2e-coverage.test.ts
@@ -83,6 +83,8 @@ const EXCLUSIONS: Readonly<Record<string, string>> = {
     "Acts on a deployment, which only `composer deploy` can create here. Same reason as `service deployment start`.",
   "service deployment delete":
     "Deletes a deployment, which only `composer deploy` can create here. Same reason as `service deployment start`.",
+  "service logs":
+    "Reads a deployment's log output, so it needs a deployment that has run. Only `composer deploy` produces one, which this suite cannot run.",
 };
 
 /**
diff --git a/packages/cli/tests/mount-coverage.test.ts b/packages/cli/tests/mount-coverage.test.ts
index 9024bf63..b67bbf73 100644
--- a/packages/cli/tests/mount-coverage.test.ts
+++ b/packages/cli/tests/mount-coverage.test.ts
@@ -162,6 +162,7 @@ const EXPECTED_MOUNT_PATHS: readonly string[] = [
   "service domain show",
   "service domain wait",
   "service list",
+  "service logs",
   "service open",
   "service remove",
   "service show",
diff --git a/packages/cli/tests/service-logs.test.ts b/packages/cli/tests/service-logs.test.ts
new file mode 100644
index 00000000..c5444da0
--- /dev/null
+++ b/packages/cli/tests/service-logs.test.ts
@@ -0,0 +1,596 @@
+import { describe, expect, it } from "vitest";
+
+import {
+  makeServiceCli,
+  page,
+  type Routes,
+  readFlowRoutes,
+  SERVICE,
+  SERVICE_DETAIL,
+} from "./service-testkit";
+
+type LogRecord =
+  | { type: "log"; text: string; byteStart: number; byteEnd: number }
+  | {
+      type: "terminal";
+      kind: "end" | "error";
+      code: string;
+      message: string;
+      retryable: boolean;
+      cursor: string | null;
+    };
+
+/** Chunks that ignore record boundaries: every record is cut in half and
+ *  the last one carries no trailing newline, so each page drives both
+ *  the reader's partial-line buffer and its end-of-stream tail. */
+function ndjsonStream(records: LogRecord[]): ReadableStream<Uint8Array> {
+  const encoder = new TextEncoder();
+  const chunks = records.flatMap((record, index) => {
+    const line = JSON.stringify(record);
+    const split = Math.floor(line.length / 2);
+    return [
+      line.slice(0, split),
+      line.slice(split) + (index === records.length - 1 ? "" : "\n"),
+    ];
+  });
+  return new ReadableStream({
+    start(controller) {
+      for (const chunk of chunks) {
+        controller.enqueue(encoder.encode(chunk));
+      }
+      controller.close();
+    },
+  });
+}
+
+function log(text: string, byteStart = 0): LogRecord {
+  return { type: "log", text, byteStart, byteEnd: byteStart + text.length };
+}
+
+function end(cursor: string | null): LogRecord {
+  return {
+    type: "terminal",
+    kind: "end",
+    code: "end",
+    message: "page complete",
+    retryable: false,
+    cursor,
+  };
+}
+
+function errorTerminal(retryable: boolean): LogRecord {
+  return {
+    type: "terminal",
+    kind: "error",
+    code: "log_read_failed",
+    message: "The log store is unavailable.",
+    retryable,
+    cursor: null,
+  };
+}
+
+/** One page per request, in order; the last one repeats once exhausted.
+ *  Every request's query is captured so a test can assert what it sent. */
+function logRoutes(
+  pages: LogRecord[][],
+  queries: Array<Record<string, unknown> | undefined>,
+  onRequest?: (index: number) => void,
+): Routes {
+  let index = 0;
+  return readFlowRoutes({
+    "GET /v1/deployments/{deploymentId}/logs": (init) => {
+      queries.push(init.params?.query);
+      onRequest?.(index);
+      const records = pages[Math.min(index, pages.length - 1)] ?? [];
+      index += 1;
+      return { data: ndjsonStream(records) };
+    },
+  });
+}
+
+function outputs(events: readonly { kind: string }[]) {
+  return events
+    .filter((event) => event.kind === "output")
+    .map((event) => {
+      const output = event as unknown as { channel: string; line: string };
+      return { channel: output.channel, line: output.line };
+    });
+}
+
+function dataLines(events: readonly { kind: string }[]): string[] {
+  return outputs(events)
+    .filter((output) => output.channel === "data")
+    .map((output) => output.line);
+}
+
+const TARGET = ["--project", "acme-app", "--service", "hello-world"];
+/** Polling is instant so a follow test does not wait on the 2s default. */
+const FAST_POLL = { PRISMA_CLI_SERVICE_LOGS_POLL_MS: "0" };
+
+describe("prisma-cli service logs", () => {
+  it("reads one page of the live deployment's logs and exits 0", async () => {
+    const queries: Array<Record<string, unknown> | undefined> = [];
+    const harness = await makeServiceCli({
+      routes: logRoutes(
+        [[log("first line"), log("second line"), end("42")]],
+        queries,
+      ),
+    });
+
+    const result = await harness.cli.run(["service", "logs", ...TARGET], {
+      cwd: harness.cwd,
+      env: harness.env,
+    });
+
+    expect(result.exitCode).toBe(0);
+    // The service's latest deployment is dep_2, so that is what is read.
+    expect(outputs(result.events)).toContainEqual({
+      channel: "diagnostic",
+      line: "deployment: dep_2",
+    });
+    expect(dataLines(result.events)).toEqual(["first line", "second line"]);
+    // One page only: no follow, so no second request.
+    expect(queries).toHaveLength(1);
+    // The endpoint's own default, sent explicitly.
+    expect(queries[0]).toEqual({ tail: 100 });
+    // The routine terminal record ends the page and is not printed.
+    expect(outputs(result.events)).not.toContainEqual(
+      expect.objectContaining({ line: "page complete" }),
+    );
+  });
+
+  it("passes --tail through as the page size", async () => {
+    const queries: Array<Record<string, unknown> | undefined> = [];
+    const harness = await makeServiceCli({
+      routes: logRoutes([[log("only"), end("1")]], queries),
+    });
+
+    const result = await harness.cli.run(
+      ["service", "logs", "--tail", "500", ...TARGET],
+      { cwd: harness.cwd, env: harness.env },
+    );
+
+    expect(result.exitCode).toBe(0);
+    expect(queries[0]).toEqual({ tail: 500 });
+  });
+
+  it("maps --from-start to from_start and sends no tail", async () => {
+    const queries: Array<Record<string, unknown> | undefined> = [];
+    const harness = await makeServiceCli({
+      routes: logRoutes([[log("beginning"), end("1")]], queries),
+    });
+
+    const result = await harness.cli.run(
+      ["service", "logs", "--from-start", ...TARGET],
+      { cwd: harness.cwd, env: harness.env },
+    );
+
+    expect(result.exitCode).toBe(0);
+    expect(queries[0]).toEqual({ from_start: "true" });
+  });
+
+  it("refuses --tail together with --from-start before reading anything", async () => {
+    const queries: Array<Record<string, unknown> | undefined> = [];
+    const harness = await makeServiceCli({
+      routes: logRoutes([[log("never read"), end(null)]], queries),
+    });
+
+    const result = await harness.cli.run(
+      ["service", "logs", "--tail", "5", "--from-start", ...TARGET, "--json"],
+      { cwd: harness.cwd, env: harness.env },
+    );
+
+    expect(result.exitCode).toBe(2);
+    const frame = result.json[result.json.length - 1];
+    if (frame?.kind !== "result" || frame.envelope.ok) {
+      throw new Error("expected an errored envelope");
+    }
+    expect(frame.envelope.error.code).toBe("SERVICE.LOGS_RANGE_CONFLICT");
+    // Refused before any work: the target was never resolved or read.
+    expect(queries).toEqual([]);
+  });
+
+  it("reads an explicit --deployment resolved globally", async () => {
+    const queries: Array<Record<string, unknown> | undefined> = [];
+    const harness = await makeServiceCli({
+      routes: logRoutes([[log("from dep_1"), end("7")]], queries),
+    });
+
+    const result = await harness.cli.run(
+      ["service", "logs", "--deployment", "dep_1", "--project", "acme-app"],
+      { cwd: harness.cwd, env: harness.env },
+    );
+
+    expect(result.exitCode).toBe(0);
+    expect(outputs(result.events)).toContainEqual({
+      channel: "diagnostic",
+      line: "deployment: dep_1",
+    });
+    expect(dataLines(result.events)).toEqual(["from dep_1"]);
+  });
+
+  it("settles an unknown --deployment as SERVICE.DEPLOYMENT_NOT_FOUND", async () => {
+    const harness = await makeServiceCli({
+      routes: logRoutes([[end(null)]], []),
+    });
+
+    const result = await harness.cli.run(
+      [
+        "service",
+        "logs",
+        "--deployment",
+        "dep_missing",
+        "--project",
+        "acme-app",
+        "--json",
+      ],
+      { cwd: harness.cwd, env: harness.env },
+    );
+
+    expect(result.exitCode).toBe(2);
+    const frame = result.json[result.json.length - 1];
+    if (frame?.kind !== "result" || frame.envelope.ok) {
+      throw new Error("expected an errored envelope");
+    }
+    expect(frame.envelope.error.code).toBe("SERVICE.DEPLOYMENT_NOT_FOUND");
+  });
+
+  it("settles a deployment owned by another project as its own failure", async () => {
+    const queries: Array<Record<string, unknown> | undefined> = [];
+    const harness = await makeServiceCli({
+      routes: {
+        ...logRoutes([[end(null)]], queries),
+        // The deployment's owning service is found by the global scan
+        // (no branch scope), but the resolved project's own listing is
+        // branch-scoped and does not contain it.
+        "GET /v1/apps": (init) => ({
+          data: init.params?.query?.branchGitName ? page([]) : page([SERVICE]),
+        }),
+      },
+    });
+
+    const result = await harness.cli.run(
+      [
+        "service",
+        "logs",
+        "--deployment",
+        "dep_1",
+        "--project",
+        "acme-app",
+        "--json",
+      ],
+      { cwd: harness.cwd, env: harness.env },
+    );
+
+    expect(result.exitCode).toBe(2);
+    const frame = result.json[result.json.length - 1];
+    if (frame?.kind !== "result" || frame.envelope.ok) {
+      throw new Error("expected an errored envelope");
+    }
+    expect(frame.envelope.error.code).toBe(
+      "SERVICE.DEPLOYMENT_OUTSIDE_PROJECT",
+    );
+    expect(queries).toEqual([]);
+  });
+
+  it("settles a service with no live deployment as SERVICE.NO_DEPLOYMENTS", async () => {
+    const harness = await makeServiceCli({
+      routes: {
+        ...logRoutes([[end(null)]], []),
+        "GET /v1/apps": () => ({
+          data: page([{ ...SERVICE, latestDeploymentId: null }]),
+        }),
+        "GET /v1/apps/{appId}": () => ({
+          data: { data: { ...SERVICE_DETAIL, latestDeploymentId: null } },
+        }),
+      },
+    });
+
+    const result = await harness.cli.run(
+      ["service", "logs", ...TARGET, "--json"],
+      { cwd: harness.cwd, env: harness.env },
+    );
+
+    expect(result.exitCode).toBe(2);
+    const frame = result.json[result.json.length - 1];
+    if (frame?.kind !== "result" || frame.envelope.ok) {
+      throw new Error("expected an errored envelope");
+    }
+    expect(frame.envelope.error.code).toBe("SERVICE.NO_DEPLOYMENTS");
+  });
+
+  it("settles an error terminal record as a structured failure", async () => {
+    const harness = await makeServiceCli({
+      routes: logRoutes(
+        [[log("before the failure"), errorTerminal(false)]],
+        [],
+      ),
+    });
+
+    const result = await harness.cli.run(
+      ["service", "logs", ...TARGET, "--json"],
+      { cwd: harness.cwd, env: harness.env },
+    );
+
+    expect(result.exitCode).toBe(2);
+    const frame = result.json[result.json.length - 1];
+    if (frame?.kind !== "result" || frame.envelope.ok) {
+      throw new Error("expected an errored envelope");
+    }
+    expect(frame.envelope.error.code).toBe("SERVICE.LOGS_FAILED");
+    expect(frame.envelope.error.why).toBe("The log store is unavailable.");
+    expect(frame.envelope.error.meta).toMatchObject({
+      code: "log_read_failed",
+      retryable: false,
+    });
+  });
+
+  /**
+   * A page always ends with a terminal record, so a body that stops
+   * without one was cut short. Exiting 0 there would present a partial
+   * log as the whole page, with nothing telling the user it was not.
+   */
+  it("settles a truncated page as SERVICE.LOGS_INCOMPLETE, keeping the lines it read", async () => {
+    const harness = await makeServiceCli({
+      routes: logRoutes([[log("arrived"), log("also arrived")]], []),
+    });
+
+    const result = await harness.cli.run(
+      ["service", "logs", ...TARGET, "--json"],
+      { cwd: harness.cwd, env: harness.env },
+    );
+
+    expect(result.exitCode).toBe(2);
+    const frame = result.json[result.json.length - 1];
+    if (frame?.kind !== "result" || frame.envelope.ok) {
+      throw new Error("expected an errored envelope");
+    }
+    expect(frame.envelope.error.code).toBe("SERVICE.LOGS_INCOMPLETE");
+    // What did arrive is still reported: the refusal is about
+    // completeness, not about discarding the lines that were read.
+    expect(dataLines(result.events)).toEqual(["arrived", "also arrived"]);
+  });
+
+  it("settles a refused request as SERVICE.LOGS_FAILED carrying the status", async () => {
+    const harness = await makeServiceCli({
+      routes: readFlowRoutes({
+        "GET /v1/deployments/{deploymentId}/logs": () => ({
+          error: { error: { message: "boom" } },
+          status: 500,
+        }),
+      }),
+    });
+
+    const result = await harness.cli.run(
+      ["service", "logs", ...TARGET, "--json"],
+      { cwd: harness.cwd, env: harness.env },
+    );
+
+    expect(result.exitCode).toBe(2);
+    const frame = result.json[result.json.length - 1];
+    if (frame?.kind !== "result" || frame.envelope.ok) {
+      throw new Error("expected an errored envelope");
+    }
+    expect(frame.envelope.error.code).toBe("SERVICE.LOGS_FAILED");
+    expect(frame.envelope.error.meta).toMatchObject({ status: 500 });
+  });
+
+  it("frames log lines as json stream events, carrying their byte range", async () => {
+    const harness = await makeServiceCli({
+      routes: logRoutes([[log("framed", 10), end("1")]], []),
+    });
+
+    const result = await harness.cli.run(
+      ["service", "logs", ...TARGET, "--json"],
+      { cwd: harness.cwd, env: harness.env },
+    );
+
+    expect(result.exitCode).toBe(0);
+    // Session output is framed per record, so a json consumer reads the
+    // line and the byte range it covered.
+    expect(result.json).toContainEqual(
+      expect.objectContaining({
+        kind: "output",
+        channel: "data",
+        line: "framed",
+        data: { byteStart: 10, byteEnd: 16 },
+      }),
+    );
+    const last = result.json[result.json.length - 1];
+    if (last?.kind !== "result" || !last.envelope.ok) {
+      throw new Error("expected a completed envelope");
+    }
+    expect(last.envelope.commandId).toBe("service.logs");
+  });
+
+  it("fails early with the engine sign-in error when unauthenticated", async () => {
+    const harness = await makeServiceCli({
+      routes: logRoutes([[end(null)]], []),
+      authenticated: false,
+    });
+
+    const result = await harness.cli.run(["service", "logs", ...TARGET], {
+      cwd: harness.cwd,
+      env: harness.env,
+      isTty: { stdout: true },
+    });
+
+    expect(result.exitCode).toBe(2);
+    expect(result.stderr).toContain("CLI.CREDENTIALS_REQUIRED");
+  });
+});
+
+describe("prisma-cli service logs --follow", () => {
+  it("polls from the cursor the previous page ended on until interrupted", async () => {
+    const queries: Array<Record<string, unknown> | undefined> = [];
+    const controller = new AbortController();
+    const harness = await makeServiceCli({
+      routes: logRoutes(
+        [
+          [log("page one"), end("100")],
+          [log("page two"), end("200")],
+        ],
+        queries,
+        // Interrupting from the fixture keeps the loop deterministic:
+        // the run ends after exactly two pages, not after a timer.
+        (index) => {
+          if (index === 1) {
+            controller.abort("SIGINT");
+          }
+        },
+      ),
+    });
+
+    const result = await harness.cli.run(
+      ["service", "logs", "--follow", ...TARGET],
+      {
+        cwd: harness.cwd,
+        env: { ...harness.env, ...FAST_POLL },
+        abort: controller.signal,
+      },
+    );
+
+    // The engine settles an interrupted session at 128 + SIGINT.
+    expect(result.exitCode).toBe(130);
+    expect(dataLines(result.events)).toEqual(["page one", "page two"]);
+    // First page asks for the tail; each later page resumes from the
+    // cursor the one before it ended on.
+    expect(queries[0]).toEqual({ tail: 100 });
+    expect(queries[1]).toEqual({ cursor: "100" });
+  });
+
+  /**
+   * The retry budget is per failure, not per run: a page that succeeds
+   * restores it. Without that reset the second retryable error below
+   * would end the run, so this fixture is what separates "one retry per
+   * failure" from "one retry ever".
+   */
+  it("recovers from a retryable error and can retry again later", async () => {
+    const queries: Array<Record<string, unknown> | undefined> = [];
+    const controller = new AbortController();
+    const harness = await makeServiceCli({
+      routes: logRoutes(
+        [
+          [log("a"), end("100")],
+          [errorTerminal(true)],
+          [log("b"), end("200")],
+          [errorTerminal(true)],
+          [log("c"), end("300")],
+        ],
+        queries,
+        (index) => {
+          if (index === 4) {
+            controller.abort("SIGINT");
+          }
+        },
+      ),
+    });
+
+    const result = await harness.cli.run(
+      ["service", "logs", "--follow", ...TARGET],
+      {
+        cwd: harness.cwd,
+        env: { ...harness.env, ...FAST_POLL },
+        abort: controller.signal,
+      },
+    );
+
+    expect(result.exitCode).toBe(130);
+    expect(dataLines(result.events)).toEqual(["a", "b", "c"]);
+  });
+
+  it("stops with SERVICE.LOGS_NO_CURSOR when a page leaves nothing to resume from", async () => {
+    const queries: Array<Record<string, unknown> | undefined> = [];
+    const harness = await makeServiceCli({
+      // A terminal record carrying no cursor: re-requesting would fall
+      // back to the default tail and reprint these same lines forever.
+      routes: logRoutes([[log("only page"), end(null)]], queries),
+    });
+
+    const result = await harness.cli.run(
+      ["service", "logs", "--follow", ...TARGET, "--json"],
+      { cwd: harness.cwd, env: { ...harness.env, ...FAST_POLL } },
+    );
+
+    expect(result.exitCode).toBe(2);
+    const frame = result.json[result.json.length - 1];
+    if (frame?.kind !== "result" || frame.envelope.ok) {
+      throw new Error("expected an errored envelope");
+    }
+    expect(frame.envelope.error.code).toBe("SERVICE.LOGS_NO_CURSOR");
+    // The page itself was read and printed; only the follow stops.
+    expect(dataLines(result.events)).toEqual(["only page"]);
+    // And it stopped before asking for anything a second time.
+    expect(queries).toHaveLength(1);
+  });
+
+  /** A truncated page is an incomplete read, not a page that closed with
+   *  nothing to resume from, so it reports the more specific failure —
+   *  the same one page mode reports for the same body. */
+  it("stops with SERVICE.LOGS_INCOMPLETE when a page carries no terminal record", async () => {
+    const queries: Array<Record<string, unknown> | undefined> = [];
+    const harness = await makeServiceCli({
+      routes: logRoutes([[log("truncated")]], queries),
+    });
+
+    const result = await harness.cli.run(
+      ["service", "logs", "--follow", ...TARGET, "--json"],
+      { cwd: harness.cwd, env: { ...harness.env, ...FAST_POLL } },
+    );
+
+    expect(result.exitCode).toBe(2);
+    const frame = result.json[result.json.length - 1];
+    if (frame?.kind !== "result" || frame.envelope.ok) {
+      throw new Error("expected an errored envelope");
+    }
+    expect(frame.envelope.error.code).toBe("SERVICE.LOGS_INCOMPLETE");
+    expect(queries).toHaveLength(1);
+  });
+
+  it("retries a retryable error terminal once, then reports it", async () => {
+    const queries: Array<Record<string, unknown> | undefined> = [];
+    const harness = await makeServiceCli({
+      routes: logRoutes(
+        [
+          [log("page one"), end("100")],
+          [errorTerminal(true)],
+          [errorTerminal(true)],
+        ],
+        queries,
+      ),
+    });
+
+    const result = await harness.cli.run(
+      ["service", "logs", "--follow", ...TARGET, "--json"],
+      { cwd: harness.cwd, env: { ...harness.env, ...FAST_POLL } },
+    );
+
+    expect(result.exitCode).toBe(2);
+    const frame = result.json[result.json.length - 1];
+    if (frame?.kind !== "result" || frame.envelope.ok) {
+      throw new Error("expected an errored envelope");
+    }
+    expect(frame.envelope.error.code).toBe("SERVICE.LOGS_FAILED");
+    // Three requests: the first page, the error, and the one retry. A
+    // fourth would mean the loop was hammering a persistent failure.
+    expect(queries).toHaveLength(3);
+  });
+
+  it("reports a non-retryable error terminal without retrying", async () => {
+    const queries: Array<Record<string, unknown> | undefined> = [];
+    const harness = await makeServiceCli({
+      routes: logRoutes(
+        [[log("page one"), end("100")], [errorTerminal(false)]],
+        queries,
+      ),
+    });
+
+    const result = await harness.cli.run(
+      ["service", "logs", "--follow", ...TARGET, "--json"],
+      { cwd: harness.cwd, env: { ...harness.env, ...FAST_POLL } },
+    );
+
+    expect(result.exitCode).toBe(2);
+    expect(queries).toHaveLength(2);
+  });
+});