diff --git a/agentkit/toolkit/cli/cli.py b/agentkit/toolkit/cli/cli.py
index 25261c2..57a7bac 100644
--- a/agentkit/toolkit/cli/cli.py
+++ b/agentkit/toolkit/cli/cli.py
@@ -47,6 +47,7 @@
 from agentkit.toolkit.cli.cli_add import add_app
 from agentkit.toolkit.cli.cli_list import list_app
 from agentkit.toolkit.cli.cli_delete import delete_app
+from agentkit.toolkit.cli.cli_logs import logs_command
 
 # Note: Avoid importing heavy packages at the top to keep CLI startup fast
 
@@ -109,6 +110,7 @@ def main(
 app.command(name="launch")(launch_command)
 app.command(name="status")(status_command)
 app.command(name="destroy")(destroy_command)
+app.command(name="logs")(logs_command)
 
 # Auth: top-level convenience commands + an `auth` group for profiles.
 app.command(name="login")(login_command)
diff --git a/agentkit/toolkit/cli/cli_logs.py b/agentkit/toolkit/cli/cli_logs.py
new file mode 100644
index 0000000..6a75911
--- /dev/null
+++ b/agentkit/toolkit/cli/cli_logs.py
@@ -0,0 +1,253 @@
+# Copyright (c) 2026 Beijing Volcano Engine Technology Co., Ltd. and/or its affiliates.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""AgentKit CLI - ``logs`` command.
+
+Query a deployed harness runtime's logs from APMPlus / TLS. The ``--harness``
+value names a runtime; it must carry the harness tag stamped at deploy time
+(``agentkit:agenttype=harness``) — otherwise it is a regular agent app whose logs
+this command cannot query.
+"""
+
+import datetime
+import re
+import time
+from pathlib import Path
+from typing import Optional
+
+import typer
+from rich.console import Console
+
+console = Console()
+
+_DURATION_UNITS_MS = {"s": 1000, "m": 60_000, "h": 3_600_000, "d": 86_400_000}
+_DURATION_RE = re.compile(r"(\d+)([smhd])")
+
+
+def _parse_since(since: str) -> int:
+    """Parse a relative duration like ``1h`` / ``30m`` / ``1h30m`` / ``2d`` to ms.
+
+    Raises:
+        ValueError: when the string has no recognizable ``<number><unit>`` token.
+    """
+    matches = _DURATION_RE.findall(since.strip().lower())
+    if not matches or "".join(n + u for n, u in matches) != since.strip().lower():
+        raise ValueError(
+            f"无法解析 --since '{since}'，请使用形如 1h / 30m / 2d / 1h30m 的格式。"
+        )
+    return sum(int(n) * _DURATION_UNITS_MS[u] for n, u in matches)
+
+
+def _format_timestamp(ts) -> str:
+    """Format a millisecond epoch timestamp to local time; pass through on failure."""
+    try:
+        return datetime.datetime.fromtimestamp(int(ts) / 1000).strftime(
+            "%Y-%m-%d %H:%M:%S"
+        )
+    except (ValueError, TypeError, OSError):
+        return str(ts)
+
+
+def _render_line(entry: dict) -> str:
+    """Render one log entry as a plain ``<time> [level] <message>`` line."""
+    ts = _format_timestamp(entry.get("__time__", ""))
+    # AgentKit runtime logs carry the line in `message`; fall back to the generic
+    # TLS `__content__` field for other topics.
+    content = entry.get("message") or entry.get("__content__") or ""
+    level = entry.get("log_level")
+    return f"{ts} {level} {content}" if level else f"{ts} {content}"
+
+
+def _find_runtime(client, target: str):
+    """Return the runtime whose name (or RuntimeId) equals ``target``, or None."""
+    from agentkit.sdk.runtime import types as rt
+
+    next_token = None
+    while True:
+        resp = client.list_runtimes(
+            rt.ListRuntimesRequest(max_results=50, next_token=next_token)
+        )
+        for runtime in resp.agent_kit_runtimes or []:
+            if runtime.name == target or runtime.runtime_id == target:
+                return runtime
+        next_token = resp.next_token
+        if not next_token:
+            return None
+
+
+def logs_command(
+    harness: str = typer.Option(
+        ...,
+        "--harness",
+        help="Harness runtime name (or RuntimeId) to query logs for.",
+    ),
+    region: Optional[str] = typer.Option(
+        None, "--region", help="Region override (default: cn-beijing / global config)."
+    ),
+    query: Optional[str] = typer.Option(
+        None,
+        "--query",
+        help="Override the TLS query. Default: service:<runtime_id>.<name>.",
+    ),
+    since: Optional[str] = typer.Option(
+        None,
+        "--since",
+        help="Relative window from now, e.g. 1h / 30m / 2d / 1h30m. Conflicts with --start.",
+    ),
+    start_time: Optional[int] = typer.Option(
+        None, "--start", help="Query start time (epoch ms). Default: 15 minutes ago."
+    ),
+    end_time: Optional[int] = typer.Option(
+        None, "--end", help="Query end time (epoch ms). Default: now."
+    ),
+    limit: int = typer.Option(
+        200, "--limit", help="Max log entries to return (default: 200)."
+    ),
+    sort: str = typer.Option(
+        "desc", "--sort", help="Order by time: desc (newest first) or asc."
+    ),
+    tls_endpoint: Optional[str] = typer.Option(
+        None, "--tls-endpoint", help="TLS host override (default: tls-<region>.volces.com)."
+    ),
+    output: Optional[Path] = typer.Option(
+        None,
+        "--output",
+        "-o",
+        help="Write logs to this file (parent dirs created) instead of stdout.",
+    ),
+    raw: bool = typer.Option(False, "--raw", help="Print the raw SearchLogs JSON."),
+) -> None:
+    """Query a deployed harness runtime's logs (APMPlus / TLS).
+
+    Examples:
+        agentkit logs --harness research-agent
+        agentkit logs --harness my-harness --limit 50 --sort asc
+        agentkit logs --harness research-agent --since 1h --output ./logs/research-agent.log
+    """
+    import json as _json
+
+    if since is not None and start_time is not None:
+        console.print("[red]❌ --since 与 --start 不能同时使用。[/red]")
+        raise typer.Exit(1)
+
+    from agentkit.platform import VolcConfiguration, resolve_credentials
+    from agentkit.sdk.runtime.client import AgentkitRuntimeClient
+    from agentkit.toolkit.harness.deploy import HARNESS_TAG_KEY, HARNESS_TAG_VALUE
+    from agentkit.toolkit.volcengine import apmplus_logs
+
+    cfg = VolcConfiguration(region=region or None)
+    resolved_region = cfg.region
+    try:
+        creds = resolve_credentials("agentkit", platform_config=cfg)
+    except ValueError as exc:
+        console.print(f"[red]❌ {exc}[/red]")
+        raise typer.Exit(1)
+
+    # Resolve the runtime and confirm it is a harness before doing anything else.
+    client = AgentkitRuntimeClient(region=resolved_region)
+    with console.status("[cyan]Resolving runtime...[/cyan]", spinner="dots"):
+        runtime = _find_runtime(client, harness)
+
+    if runtime is None:
+        console.print(
+            f"[red]❌ 未找到名为 '{harness}' 的运行时（region: {resolved_region}）。[/red]"
+        )
+        raise typer.Exit(1)
+
+    is_harness = any(
+        tag.key == HARNESS_TAG_KEY and tag.value == HARNESS_TAG_VALUE
+        for tag in (runtime.tags or [])
+    )
+    if not is_harness:
+        console.print("[red]❌ 非 Harness 应用，无法查询日志[/red]")
+        raise typer.Exit(1)
+
+    runtime_id = runtime.runtime_id or ""
+    runtime_name = runtime.name or harness
+    final_query = query or f"service:{runtime_id}.{runtime_name}"
+
+    now_ms = int(time.time() * 1000)
+    end_ms = end_time if end_time is not None else now_ms
+    if since is not None:
+        try:
+            start_ms = end_ms - _parse_since(since)
+        except ValueError as exc:
+            console.print(f"[red]❌ {exc}[/red]")
+            raise typer.Exit(1)
+    elif start_time is not None:
+        start_ms = start_time
+    else:
+        start_ms = end_ms - 15 * 60 * 1000
+
+    try:
+        with console.status("[cyan]Fetching log topic...[/cyan]", spinner="dots"):
+            topic_id = apmplus_logs.get_log_topic_id(
+                access_key=creds.access_key,
+                secret_key=creds.secret_key,
+                region=resolved_region,
+                session_token=creds.session_token,
+            )
+        with console.status("[cyan]Searching logs...[/cyan]", spinner="dots"):
+            response = apmplus_logs.search_logs(
+                access_key=creds.access_key,
+                secret_key=creds.secret_key,
+                region=resolved_region,
+                topic_id=topic_id,
+                query=final_query,
+                start_time_ms=start_ms,
+                end_time_ms=end_ms,
+                limit=limit,
+                sort=sort,
+                session_token=creds.session_token,
+                tls_host=tls_endpoint,
+            )
+    except ValueError as exc:
+        console.print(f"[red]❌ 查询日志失败: {exc}[/red]")
+        raise typer.Exit(1)
+
+    entries = apmplus_logs.flatten_logs(response)
+
+    # Build the text payload once (raw JSON or rendered lines), then either write
+    # it to --output or print it to the console.
+    if raw:
+        payload = _json.dumps(response, ensure_ascii=False, indent=2)
+    else:
+        payload = "\n".join(_render_line(entry) for entry in entries)
+
+    if output is not None:
+        output.parent.mkdir(parents=True, exist_ok=True)
+        output.write_text(payload + ("\n" if payload else ""), encoding="utf-8")
+        console.print(
+            f"[green]✅ 命中 {len(entries)} 条日志，已写入 {output}[/green]"
+        )
+        return
+
+    if raw:
+        console.print(payload)
+        return
+
+    console.print(f"[blue]Query: {final_query}[/blue]")
+    if not entries:
+        console.print("[yellow]未查询到日志。[/yellow]")
+        return
+
+    console.print(f"[green]✅ 命中 {len(entries)} 条日志[/green]")
+    for entry in entries:
+        ts = _format_timestamp(entry.get("__time__", ""))
+        content = entry.get("message") or entry.get("__content__") or ""
+        level = entry.get("log_level")
+        prefix = f"[magenta]{ts}[/magenta]"
+        if level:
+            prefix += f" [yellow]{level}[/yellow]"
+        console.print(f"{prefix} {content}")
diff --git a/agentkit/toolkit/executors/init_executor.py b/agentkit/toolkit/executors/init_executor.py
index 6509f55..b24a95f 100644
--- a/agentkit/toolkit/executors/init_executor.py
+++ b/agentkit/toolkit/executors/init_executor.py
@@ -143,8 +143,10 @@
 # Container image for the harness server. The base image's apt mirror is an
 # unreachable internal host, so apt is repointed at aliyun; the source branch is
 # cloned via the ghfast proxy with a github fallback; uv installs from aliyun.
-# `openai-codex` is installed alongside veadk so the `codex` runtime works
-# (it bundles the Codex CLI binary); without it `--runtime codex` fails.
+# `openai-codex` is installed so the `codex` runtime works (it bundles the Codex
+# CLI binary); without it `--runtime codex` fails. It is pre-release only, so it
+# needs --prerelease=allow and is pulled from the Volcengine mirror (which
+# carries the pre-release wheels and is fastest from the build infra).
 _HARNESS_DOCKERFILE = """\
 FROM agentkit-cn-beijing.cr.volces.com/base/py-simple:python3.12-bookworm-slim-latest
 ENV PYTHONUNBUFFERED=1
@@ -168,7 +170,13 @@
     done; \\
     test -d src/veadk
 RUN uv pip install --system --index-url https://mirrors.aliyun.com/pypi/simple/ \\
-        ./src fastapi "uvicorn[standard]" openai-codex
+        ./src fastapi "uvicorn[standard]"
+# Codex runtime (RUNTIME=codex): openai-codex + its bundled CLI binary are
+# pre-release only, so --prerelease=allow is required; pulled from the
+# Volcengine mirror which carries the pre-release wheels.
+RUN uv pip install --system --prerelease=allow \\
+        --index-url https://mirrors.volces.com/pypi/simple/ \\
+        openai-codex
 EXPOSE 8000
 CMD ["python", "-m", "uvicorn", "veadk.cloud.harness_app.app:app", "--host", "0.0.0.0", "--port", "8000"]
 """
diff --git a/agentkit/toolkit/volcengine/apmplus_logs.py b/agentkit/toolkit/volcengine/apmplus_logs.py
new file mode 100644
index 0000000..f39290c
--- /dev/null
+++ b/agentkit/toolkit/volcengine/apmplus_logs.py
@@ -0,0 +1,184 @@
+# Copyright (c) 2026 Beijing Volcano Engine Technology Co., Ltd. and/or its affiliates.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Query a harness runtime's logs from APMPlus / TLS (VE-sign auth).
+
+Two steps:
+
+1. ``GetLogModuleConfig`` (APMPlus OpenAPI, Action style) — signed with
+   Volcengine SigV4 via :func:`agentkit.auth._sigv4.sign_headers` — returns the
+   TLS log topic id that AgentKit runtime logs are written to.
+2. ``SearchLogs`` (TLS native API) — issued via the official ``volcengine.tls``
+   SDK, which handles TLS's own SigV4 variant — searches that topic.
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Any, Dict, List, Optional
+
+import requests
+
+from agentkit.auth._sigv4 import sign_headers
+
+logger = logging.getLogger("agentkit." + __name__)
+
+# APMPlus OpenAPI: returns the runtime log module config (incl. the TLS topic).
+_APMPLUS_SERVICE = "apmplus_server"
+_APMPLUS_ACTION = "GetLogModuleConfig"
+_APMPLUS_VERSION = "2024-07-30"
+
+_TIMEOUT = 30
+
+
+def _apmplus_host(region: str) -> str:
+    return f"apmplus-server.{region}.volcengineapi.com"
+
+
+def _tls_host(region: str) -> str:
+    return f"tls-{region}.volces.com"
+
+
+def get_log_topic_id(
+    *,
+    access_key: str,
+    secret_key: str,
+    region: str,
+    session_token: Optional[str] = None,
+    scheme: str = "https",
+) -> str:
+    """Resolve the TLS log topic id via APMPlus ``GetLogModuleConfig``.
+
+    The action takes no request parameters; it returns the account/region log
+    module config, whose ``LogTopicId`` is the TLS topic AgentKit runtimes log to.
+
+    Raises:
+        ValueError: on a non-200 response or when ``LogTopicId`` is absent.
+    """
+    host = _apmplus_host(region)
+    query = {"Action": _APMPLUS_ACTION, "Version": _APMPLUS_VERSION}
+    body = b"{}"
+    headers = sign_headers(
+        "POST",
+        host,
+        query,
+        body,
+        access_key=access_key,
+        secret_key=secret_key,
+        service=_APMPLUS_SERVICE,
+        region=region,
+        session_token=session_token,
+    )
+    resp = requests.post(
+        f"{scheme}://{host}/", params=query, data=body, headers=headers, timeout=_TIMEOUT
+    )
+    if resp.status_code != 200:
+        raise ValueError(
+            f"GetLogModuleConfig failed (HTTP {resp.status_code}): {resp.text}"
+        )
+
+    data = resp.json()
+    # APMPlus returns {"data": {"LogTopicId": ...}, ...}; the OpenAPI gateway may
+    # additionally wrap it under "Result". Accept either shape.
+    container = data.get("Result", data) if isinstance(data, dict) else {}
+    module = container.get("data", {}) if isinstance(container, dict) else {}
+    topic_id = module.get("LogTopicId") if isinstance(module, dict) else None
+    if not topic_id:
+        raise ValueError(f"GetLogModuleConfig returned no LogTopicId: {data}")
+    return str(topic_id)
+
+
+def search_logs(
+    *,
+    access_key: str,
+    secret_key: str,
+    region: str,
+    topic_id: str,
+    query: str,
+    start_time_ms: int,
+    end_time_ms: int,
+    limit: int = 20,
+    sort: str = "desc",
+    session_token: Optional[str] = None,
+    scheme: str = "https",
+    tls_host: Optional[str] = None,
+) -> Dict[str, Any]:
+    """Search a TLS topic via the native ``SearchLogs`` API (api-version 0.3.0).
+
+    Uses the official ``volcengine.tls`` SDK, which applies the TLS-specific
+    request signing (service ``TLS``, ``Content-MD5`` / ``x-tls-apiversion``
+    headers). Returns a normalized ``{ResultStatus, Count, Logs, ...}`` dict.
+
+    Raises:
+        ValueError: when the TLS API rejects the request.
+    """
+    from volcengine.tls.TLSService import TLSService
+    from volcengine.tls.tls_requests import SearchLogsRequest
+    from volcengine.tls.tls_exception import TLSException
+
+    endpoint = tls_host or _tls_host(region)
+    # Only pass a session token when present: the SDK treats its absence (default)
+    # as "no token", and signing an empty token would be rejected.
+    token_kwargs = {"security_token": session_token} if session_token else {}
+    service = TLSService(
+        endpoint=f"{scheme}://{endpoint}",
+        access_key_id=access_key,
+        access_key_secret=secret_key,
+        region=region,
+        timeout=_TIMEOUT,
+        **token_kwargs,
+    )
+    request = SearchLogsRequest(
+        topic_id=topic_id,
+        query=query,
+        start_time=start_time_ms,
+        end_time=end_time_ms,
+        limit=limit,
+        sort=sort,
+    )
+    try:
+        result = service.search_logs_v2(request).get_search_result()
+    except TLSException as exc:
+        raise ValueError(f"SearchLogs failed: {exc}") from exc
+
+    if result is None:
+        raise ValueError("SearchLogs returned no result")
+
+    return {
+        "ResultStatus": result.result_status,
+        "Count": result.count,
+        "Limit": result.limit,
+        "ListOver": result.list_over,
+        "Analysis": result.analysis,
+        "Context": result.context,
+        "Logs": result.logs or [],
+    }
+
+
+def flatten_logs(response: Dict[str, Any]) -> List[Dict[str, Any]]:
+    """Flatten the ``Logs`` field of a SearchLogs response to a list of dicts.
+
+    ``Logs`` is normally a list of JSON maps; for some context queries it is a
+    list of single-element lists. Both shapes are flattened to dict entries.
+    """
+    logs = response.get("Logs") if isinstance(response, dict) else None
+    if not isinstance(logs, list):
+        return []
+    out: List[Dict[str, Any]] = []
+    for item in logs:
+        if isinstance(item, dict):
+            out.append(item)
+        elif isinstance(item, list):
+            out.extend(x for x in item if isinstance(x, dict))
+    return out
diff --git a/tests/toolkit/cli/test_cli_logs.py b/tests/toolkit/cli/test_cli_logs.py
new file mode 100644
index 0000000..b92f04d
--- /dev/null
+++ b/tests/toolkit/cli/test_cli_logs.py
@@ -0,0 +1,223 @@
+# Copyright (c) 2026 Beijing Volcano Engine Technology Co., Ltd. and/or its affiliates.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Tests for ``agentkit logs --harness <name>``."""
+
+from types import SimpleNamespace
+
+from typer.testing import CliRunner
+
+from agentkit.platform.configuration import Credentials
+from agentkit.toolkit.cli.cli import app
+from agentkit.toolkit.harness.deploy import HARNESS_TAG_KEY, HARNESS_TAG_VALUE
+
+runner = CliRunner()
+
+
+def _tag(key, value):
+    return SimpleNamespace(key=key, value=value)
+
+
+def _runtime(name, runtime_id, *, harness):
+    tags = [_tag(HARNESS_TAG_KEY, HARNESS_TAG_VALUE)] if harness else []
+    return SimpleNamespace(name=name, runtime_id=runtime_id, tags=tags)
+
+
+class _FakeClient:
+    """Stand-in for AgentkitRuntimeClient returning a fixed runtime page."""
+
+    runtimes: list = []
+
+    def __init__(self, *args, **kwargs):
+        pass
+
+    def list_runtimes(self, request):
+        return SimpleNamespace(agent_kit_runtimes=self.runtimes, next_token=None)
+
+
+def _patch_common(monkeypatch, runtimes):
+    _FakeClient.runtimes = runtimes
+    monkeypatch.setattr(
+        "agentkit.sdk.runtime.client.AgentkitRuntimeClient", _FakeClient
+    )
+    monkeypatch.setattr(
+        "agentkit.platform.resolve_credentials",
+        lambda *a, **k: Credentials(access_key="ak", secret_key="sk", source="test"),
+    )
+
+
+def test_logs_non_harness_app_is_rejected(monkeypatch):
+    _patch_common(
+        monkeypatch,
+        [_runtime("drawing_assistant-1joart6y", "r-draw", harness=False)],
+    )
+    result = runner.invoke(
+        app, ["logs", "--harness", "drawing_assistant-1joart6y", "--region", "cn-beijing"]
+    )
+    assert result.exit_code == 1
+    assert "非 Harness 应用，无法查询日志" in result.stdout
+
+
+def test_logs_runtime_not_found(monkeypatch):
+    _patch_common(monkeypatch, [])
+    result = runner.invoke(
+        app, ["logs", "--harness", "missing", "--region", "cn-beijing"]
+    )
+    assert result.exit_code == 1
+    assert "未找到" in result.stdout
+
+
+def test_logs_harness_success_builds_default_query(monkeypatch):
+    _patch_common(
+        monkeypatch, [_runtime("my-harness", "r-abc", harness=True)]
+    )
+
+    captured = {}
+
+    def fake_topic(**kwargs):
+        return "topic-123"
+
+    def fake_search(**kwargs):
+        captured.update(kwargs)
+        return {
+            "Logs": [
+                {"__time__": "1781683297505", "__content__": "hello from harness"}
+            ]
+        }
+
+    monkeypatch.setattr(
+        "agentkit.toolkit.volcengine.apmplus_logs.get_log_topic_id", fake_topic
+    )
+    monkeypatch.setattr(
+        "agentkit.toolkit.volcengine.apmplus_logs.search_logs", fake_search
+    )
+
+    result = runner.invoke(
+        app, ["logs", "--harness", "my-harness", "--region", "cn-beijing"]
+    )
+    assert result.exit_code == 0, result.stdout
+    # Default query is built from runtime_id + name.
+    assert captured["query"] == "service:r-abc.my-harness"
+    assert captured["topic_id"] == "topic-123"
+    assert "hello from harness" in result.stdout
+
+
+def test_logs_query_override(monkeypatch):
+    _patch_common(monkeypatch, [_runtime("my-harness", "r-abc", harness=True)])
+
+    captured = {}
+
+    monkeypatch.setattr(
+        "agentkit.toolkit.volcengine.apmplus_logs.get_log_topic_id",
+        lambda **k: "t",
+    )
+
+    def fake_search(**kwargs):
+        captured.update(kwargs)
+        return {"Logs": []}
+
+    monkeypatch.setattr(
+        "agentkit.toolkit.volcengine.apmplus_logs.search_logs", fake_search
+    )
+
+    result = runner.invoke(
+        app,
+        ["logs", "--harness", "my-harness", "--query", "error", "--region", "cn-beijing"],
+    )
+    assert result.exit_code == 0, result.stdout
+    assert captured["query"] == "error"
+    assert "未查询到日志" in result.stdout
+
+
+def test_parse_since_durations():
+    from agentkit.toolkit.cli.cli_logs import _parse_since
+
+    assert _parse_since("30m") == 30 * 60_000
+    assert _parse_since("1h") == 3_600_000
+    assert _parse_since("2d") == 2 * 86_400_000
+    assert _parse_since("1h30m") == 3_600_000 + 30 * 60_000
+
+
+def test_parse_since_invalid():
+    import pytest
+
+    from agentkit.toolkit.cli.cli_logs import _parse_since
+
+    with pytest.raises(ValueError):
+        _parse_since("yesterday")
+
+
+def test_logs_since_sets_window(monkeypatch):
+    _patch_common(monkeypatch, [_runtime("my-harness", "r-abc", harness=True)])
+
+    captured = {}
+    monkeypatch.setattr(
+        "agentkit.toolkit.volcengine.apmplus_logs.get_log_topic_id", lambda **k: "t"
+    )
+
+    def fake_search(**kwargs):
+        captured.update(kwargs)
+        return {"Logs": []}
+
+    monkeypatch.setattr(
+        "agentkit.toolkit.volcengine.apmplus_logs.search_logs", fake_search
+    )
+
+    result = runner.invoke(
+        app, ["logs", "--harness", "my-harness", "--since", "1h", "--region", "cn-beijing"]
+    )
+    assert result.exit_code == 0, result.stdout
+    # Window spans exactly the requested duration.
+    assert captured["end_time_ms"] - captured["start_time_ms"] == 3_600_000
+
+
+def test_logs_since_conflicts_with_start(monkeypatch):
+    _patch_common(monkeypatch, [_runtime("my-harness", "r-abc", harness=True)])
+    result = runner.invoke(
+        app,
+        [
+            "logs", "--harness", "my-harness",
+            "--since", "1h", "--start", "1", "--region", "cn-beijing",
+        ],
+    )
+    assert result.exit_code == 1
+    assert "--since 与 --start 不能同时使用" in result.stdout
+
+
+def test_logs_output_writes_file(monkeypatch, tmp_path):
+    _patch_common(monkeypatch, [_runtime("my-harness", "r-abc", harness=True)])
+    monkeypatch.setattr(
+        "agentkit.toolkit.volcengine.apmplus_logs.get_log_topic_id", lambda **k: "t"
+    )
+    monkeypatch.setattr(
+        "agentkit.toolkit.volcengine.apmplus_logs.search_logs",
+        lambda **k: {
+            "Logs": [
+                {"__time__": "1781683297505", "log_level": "INFO", "message": "line one"}
+            ]
+        },
+    )
+
+    out = tmp_path / "logs" / "my-harness.log"
+    result = runner.invoke(
+        app,
+        ["logs", "--harness", "my-harness", "--output", str(out), "--region", "cn-beijing"],
+    )
+    assert result.exit_code == 0, result.stdout
+    assert out.is_file()
+    content = out.read_text(encoding="utf-8")
+    assert "line one" in content
+    assert "INFO" in content
+    # (path may be line-wrapped by rich in captured output, so just check the verb)
+    assert "已写入" in result.stdout