Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
62 changes: 51 additions & 11 deletions src/keboola_agent_cli/auto_update.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@
import subprocess
import sys
import time
from dataclasses import dataclass
from importlib.metadata import distribution
from pathlib import Path

Expand Down Expand Up @@ -47,6 +48,7 @@
prepare_kbagent_update_plan,
prepare_mcp_update_plan,
resolve_kbagent_wheel_url,
summarize_failure_tail,
)

logger = logging.getLogger(__name__)
Expand All @@ -66,6 +68,28 @@ class UpdateOutcome(enum.Enum):
FAILED = "failed"


@dataclass(frozen=True)
class UpdateAttempt:
"""One self-update subprocess run: its outcome plus *why* it ended that way.

The startup hook runs the installer with ``capture_output=True`` and used to
throw the transcript away, so a failed auto-update printed a bare
"Auto-update failed" and every bug report arrived without the one line that
explains it -- see issues #528 and #545, where the venv was left broken on
Windows and neither report could say what uv actually refused to do. The
explicit ``kbagent update`` path has always surfaced ``result.stderr``;
carrying the tail here closes that asymmetry.

Attributes:
outcome: SUCCESS / TIMEOUT / FAILED.
detail: Last actionable line of the installer transcript (empty when
there is nothing to report, e.g. on SUCCESS or TIMEOUT).
"""

outcome: UpdateOutcome
detail: str = ""


# Process-level sentinel for the auto-update flow.
#
# Bug D fix from issue #263: ``kbagent repl`` re-enters the entire CLI
Expand Down Expand Up @@ -270,7 +294,7 @@ def _should_skip() -> bool:

def _perform_update(
latest_version: str, *, command: tuple[str, ...] | None = None
) -> UpdateOutcome:
) -> UpdateAttempt:
"""Download and install the latest version.

Delegates to :func:`build_kbagent_upgrade_command` so this path stays
Expand All @@ -285,9 +309,12 @@ def _perform_update(
latest_version: The version being updated to (for logging).

Returns:
:class:`UpdateOutcome`: ``SUCCESS`` on a clean install, ``TIMEOUT`` when
the install subprocess outran :func:`get_update_timeout` (a slow git+
build -- retried next run, not a real failure), ``FAILED`` otherwise.
:class:`UpdateAttempt` carrying ``SUCCESS`` on a clean install,
``TIMEOUT`` when the install subprocess outran
:func:`get_update_timeout` (a slow git+ build -- retried next run, not a
real failure), ``FAILED`` otherwise. A FAILED attempt also carries the
last actionable line of the installer transcript so the banner can say
what went wrong instead of only that something did.
"""
# ``command`` is supplied by the startup planner. Keep the fallback only
# for direct legacy callers; it must never be used after another stage has
Expand All @@ -298,7 +325,9 @@ def _perform_update(
else:
cmd = list(command)
if cmd is None:
return UpdateOutcome.FAILED
return UpdateAttempt(
UpdateOutcome.FAILED, "no installer found on PATH (neither uv nor pip)"
)

try:
result = subprocess.run(
Expand All @@ -307,11 +336,16 @@ def _perform_update(
text=True,
timeout=get_update_timeout(),
)
return UpdateOutcome.SUCCESS if result.returncode == 0 else UpdateOutcome.FAILED
if result.returncode == 0:
return UpdateAttempt(UpdateOutcome.SUCCESS)
# uv writes its diagnostics to stderr; fall back to stdout for
# installers that do not (and so the tail is never empty by accident).
transcript = result.stderr or result.stdout
return UpdateAttempt(UpdateOutcome.FAILED, summarize_failure_tail(transcript))
except subprocess.TimeoutExpired:
return UpdateOutcome.TIMEOUT
except OSError:
return UpdateOutcome.FAILED
return UpdateAttempt(UpdateOutcome.TIMEOUT)
except OSError as exc:
return UpdateAttempt(UpdateOutcome.FAILED, str(exc))


def _re_exec() -> None:
Expand Down Expand Up @@ -608,9 +642,10 @@ def maybe_auto_update() -> None:
return

sys.stderr.write(f"Updating kbagent v{__version__} -> v{kbagent_plan.latest_version}...\n")
outcome = _perform_update(
attempt = _perform_update(
kbagent_plan.latest_version or __version__, command=kbagent_plan.command
)
outcome = attempt.outcome
if outcome is UpdateOutcome.SUCCESS:
sys.stderr.write(f"Updated to v{kbagent_plan.latest_version}. Re-launching...\n")
os.environ[ENV_UPDATED_FROM] = __version__
Expand All @@ -622,8 +657,13 @@ def maybe_auto_update() -> None:
f"{kbagent_plan.recovery_command}\n"
)
else:
# The installer's own last line is the only clue to WHY the upgrade
# failed, and on Windows a failed upgrade can leave the tool
# environment mid-swap (#528 / #545) -- so it must reach the user,
# not just the discarded subprocess buffer.
reason = f" ({attempt.detail})" if attempt.detail else ""
sys.stderr.write(
"Auto-update failed; continuing with current version. Recover with: "
f"Auto-update failed{reason}; continuing with current version. Recover with: "
f"{kbagent_plan.recovery_command}\n"
)
except Exception:
Expand Down
30 changes: 24 additions & 6 deletions src/keboola_agent_cli/services/version_service.py
Original file line number Diff line number Diff line change
Expand Up @@ -248,6 +248,26 @@ def build_kbagent_upgrade_command(
return cmd


def summarize_failure_tail(message: str | None) -> str:
"""Compress a multi-line installer transcript to its last non-empty line.

Subprocess failures embed the whole uv/pip transcript; the actionable line
(e.g. ``error: Executable already exists: kbagent``, or a Windows
``failed to remove file ... (os error 32)``) is last. Callers surface only
that tail in a one-line message and keep the full transcript for
``--json`` / ``--verbose``.

Args:
message: The captured transcript, or None.

Returns:
The last non-empty line, or ``"update failed"`` when there is nothing
to report.
"""
lines = [ln.strip() for ln in (message or "").splitlines() if ln.strip()]
return lines[-1] if lines else "update failed"


def _render_command(command: tuple[str, ...]) -> str:
"""Render argv for the current platform's interactive shell."""
if os.name == "nt":
Expand Down Expand Up @@ -958,13 +978,11 @@ def self_update(self, *, include_prerelease: bool = False) -> dict[str, Any]:
def _summarize_failure_tail(message: str | None) -> str:
"""Compress a multi-line failure message to its last non-empty line.

Subprocess failures embed the whole uv/pip transcript; the actionable
line (e.g. ``error: Executable already exists: kbagent``) is last. We
surface only that tail in the one-line summary -- the full transcript
stays in the result's ``output`` for ``--json`` / ``--verbose``.
Thin delegate to :func:`summarize_failure_tail` so the startup
auto-update hook and the explicit ``kbagent update`` path report a
failure the same way.
"""
lines = [ln.strip() for ln in (message or "").splitlines() if ln.strip()]
return lines[-1] if lines else "update failed"
return summarize_failure_tail(message)

@classmethod
def _compose_update_summary(
Expand Down
Loading