diff --git a/docs/cli-reference.md b/docs/cli-reference.md
index 45fa3c7..77f6cc4 100644
--- a/docs/cli-reference.md
+++ b/docs/cli-reference.md
@@ -495,6 +495,41 @@ Creation confirms both identities and shows the link only on the owner terminal.
### Updating
+On macOS/Linux, hide the automatic runtime notice for one explicitly selected
+stable version, or clear that preference:
+
+```bash
+session-peer update --dismiss-runtime-notice 1.3.1
+session-peer update --clear-runtime-notice-dismissal
+```
+
+The version above is an example, not a release announcement. These local-only,
+offline commands do not check, install or approve a release. They cannot be
+combined with remote/SSH options or the check option. Only the exact normalized
+version is hidden; later releases can still produce a notice, and clearing
+restores the usual behavior. Explicit local checks retain the real latest version
+and outdated result, with separate preference metadata. Remote checks, skill
+notices and the existing global opt-out are unchanged. A remote command may still
+show the calling CLI's own cached notice; its local preference is read once, not
+once per target. Remote versions and results are never hidden by that preference.
+
+The preference is stored separately from the update cache at
+`$XDG_CONFIG_HOME/session-peer/update-notices.json`, or
+`~/.config/session-peer/update-notices.json` when that variable is unset or empty.
+The directory and file must be private and safely owned; unsafe paths are refused,
+not repaired or redirected. Invalid or future-schema files hide nothing and are
+not overwritten by either command: inspect the dedicated preference file first.
+Windows does not yet support this per-version preference; ordinary updates,
+checks and the global opt-out remain available.
+
+Safety checks cover POSIX owner/mode bits, not a full ACL audit or physical-media
+durability certification. Symlinked ancestors and explicitly relative config
+paths are intentionally refused without fallback. The filesystem must support
+directory-relative I/O, file locking and directory fsync. After inspecting and
+backing up a malformed preference, an operator may explicitly remove only that
+dedicated JSON file and run the preference command again; do not remove the lock
+file while writers may be running. No automatic repair is performed.
+
For package-managed installs, use the same manager that installed the command:
```bash
@@ -518,9 +553,10 @@ session-peer update --host web-01 # push this program if version
```
Remote update compares against the **local program's version**, not the latest
-GitHub release. It pushes even if the remote version is newer; check first and
-update the local program before distributing it. The remote copy does not fetch
-from GitHub. Remote version probes inspect only
+GitHub release. It pushes only if the remote copy is missing or older, and leaves
+a newer remote version unchanged. Check first and update the local program
+before distributing it. The remote copy does not fetch from GitHub.
+Remote version probes inspect only
`~/.local/share/session-peer/session_peer.py`, not pip/pipx/uv installations;
remote update installs that standalone path and its CLI link. For a
package-managed remote CLI, upgrade it with its own manager on that host instead.
diff --git a/docs/i18n-manifest.json b/docs/i18n-manifest.json
index 39e570c..55eb227 100644
--- a/docs/i18n-manifest.json
+++ b/docs/i18n-manifest.json
@@ -15,7 +15,7 @@
"docs/ai-assistant-guide.md": "9624ec96b92fb688fae3be3c94cb3564c77985091be451fbf1f056a1af9b9d9c",
"docs/antigravity.md": "f2b5122a0ce6e6fb5bd4d8ed50ce2c8bd2dc9c98ec8ce328405cf7c2cd24229d",
"docs/architecture/agent-transports.md": "01a4e3942e87acb476595859f2e37f9653ea1e3bc19ebc68dced1adf20a6966f",
- "docs/cli-reference.md": "7b1e6e8d12069b45e12b65ebc02e2317f42ba0fe2b27e386b70d4616cd21c581",
+ "docs/cli-reference.md": "d2021243cef2dd29a48ee5d9378ded8b6e3a19e3042c3fa13899a463f700d83c",
"docs/compatibility-v1.md": "7e551aad37a369faff7e370728376763efb54c2c9e9cfd8556a0c19aabc3d103",
"docs/contracts/handoff-v1.md": "46a953010c49214bd8f969412f26713384fbe09a0ba78ebf0fef75da40d0d889",
"docs/diagnostics.md": "0949e53612eeab2cd7c7bbda498e14d71e9cb1b1189aa7722030cb8702cd8481",
diff --git a/docs/ja/cli-reference.md b/docs/ja/cli-reference.md
index 62a1d73..9e4e21a 100644
--- a/docs/ja/cli-reference.md
+++ b/docs/ja/cli-reference.md
@@ -451,6 +451,42 @@ session-peer device offer-cancel --state /private/device-state --offer-id OFFER-
### 更新
+macOS/Linux では、指定した安定版一つの自動ランタイム通知を非表示にするか、
+その設定を解除できます。
+
+```bash
+session-peer update --dismiss-runtime-notice 1.3.1
+session-peer update --clear-runtime-notice-dismissal
+```
+
+上記のバージョンは使用例であり、リリースの告知ではありません。これらはローカルの
+設定だけをオフラインで変更し、リリースの確認・インストール・承認は行いません。
+リモート/SSH オプションや確認オプションとは併用できません。正規化したバージョンが
+完全に一致する通知だけを非表示にするため、後続リリースは通知されることがあります。
+解除すると通常の動作に戻ります。明示的なローカル確認は実際の最新バージョンと
+更新の要否を保持し、通知設定は別の情報として報告します。リモート確認、スキル通知、
+既存の全通知無効化設定は変更しません。リモートコマンドでも呼び出し元 CLI 自体の
+キャッシュ通知が表示される場合があります。そのローカル設定は対象ごとではなく一度だけ
+読み取り、リモートのバージョンや結果を非表示にはしません。
+
+設定は更新キャッシュと分けて
+`$XDG_CONFIG_HOME/session-peer/update-notices.json` に保存します。環境変数が未設定か
+空の場合は `~/.config/session-peer/update-notices.json` を使います。
+ディレクトリとファイルには安全な所有権と非公開の権限が必要です。安全でないパスは
+修復や迂回をせず拒否します。設定ファイルが不正または将来のスキーマの場合は、
+通知を非表示にしません。どちらのコマンドもそのファイルを上書きしないため、
+まず専用の設定ファイルを確認してください。
+Windows はまだバージョン別設定に対応していませんが、通常の更新・確認と
+全通知無効化は引き続き利用できます。
+
+安全確認は POSIX の所有者と権限ビットが対象であり、ACL（追加のアクセス権）の
+完全な監査や物理媒体への永続化の保証ではありません。上位パスのシンボリックリンクや
+相対指定の設定パスは、代替パスに切り替えず拒否します。ファイルシステムには、
+ディレクトリ基準の入出力、ファイルロック、ディレクトリの fsync（同期）が必要です。
+不正な設定を確認・バックアップした後、運用者は専用の JSON ファイルだけを明示的に
+削除し、設定コマンドを再実行できます。書き込みが動作中の可能性がある場合は
+ロックファイルを削除しないでください。自動修復は行いません。
+
パッケージ管理されたインストールの場合は、コマンドをインストールしたのと同じマネージャーを使用します:
```bash
@@ -474,8 +510,9 @@ session-peer update --host web-01 # push this program if version
```
リモート更新は、最新の GitHub リリースではなく、**ローカルプログラムのバージョン**と
-比較します。リモートバージョンの方が新しい場合でもプッシュするため、配布する前にまず確認し、
-ローカルプログラムを更新してください。リモートのコピーが GitHub から取得することはありません。
+比較します。リモートのコピーがない場合か古い場合だけプッシュし、新しい場合は変更しません。
+配布する前にまず確認し、ローカルプログラムを更新してください。
+リモート側で GitHub からプログラムを取得することはありません。
リモートバージョンのプローブは `~/.local/share/session-peer/session_peer.py` のみを検査し、
pip/pipx/uv のインストールは検査しません。リモート更新は、そのスタンドアロンパスと CLI
リンクをインストールします。パッケージ管理されたリモート CLI の場合は、代わりにそのホスト上の
@@ -484,7 +521,7 @@ pip/pipx/uv のインストールは検査しません。リモート更新は
ローカルまたはリモートの `update` のいずれも、Claude スキルを更新しません。スタンドアロンプログラムと
スキルの両方を更新するには、目的のリリースのチェックアウトから `install.sh` を再実行してください。
ローカルの `session-peer update --check` および `session-peer update` は、自動通知で
-使用されるのと同じキャッシュも設定します。
+使用されるのと同じキャッシュにもデータを書き込みます。
### Orca が管理する Claude プロファイル
diff --git a/docs/ko/cli-reference.md b/docs/ko/cli-reference.md
index 6f5d496..0ab1b74 100644
--- a/docs/ko/cli-reference.md
+++ b/docs/ko/cli-reference.md
@@ -450,6 +450,40 @@ session-peer device offer-cancel --state /private/device-state --offer-id OFFER-
### 업데이트
+macOS/Linux에서는 지정한 안정 버전 하나의 자동 런타임 알림만 숨기거나 그 설정을
+해제할 수 있습니다.
+
+```bash
+session-peer update --dismiss-runtime-notice 1.3.1
+session-peer update --clear-runtime-notice-dismissal
+```
+
+위 버전은 사용 예시이며 릴리스 발표가 아닙니다. 두 명령은 로컬 설정만 오프라인으로
+변경하며 릴리스를 확인·설치·승인하지 않습니다. 원격/SSH 옵션이나 확인 옵션과 함께
+사용할 수 없습니다. 정규화한 버전이 정확히 일치하는 알림만 숨기므로 이후 버전의
+알림은 표시될 수 있고, 설정을 해제하면 기존 동작으로 돌아갑니다. 명시적인 로컬
+확인은 실제 최신 버전과 구버전 여부를 그대로 보고하며 알림 설정은 별도 정보로
+표시합니다. 원격 확인, 스킬 알림, 기존의 전체 알림 비활성화 설정은 바뀌지 않습니다.
+원격 명령에서도 호출한 CLI 자체의 캐시 알림은 표시될 수 있습니다. 이때 로컬 설정은
+대상별이 아닌 한 번만 읽으며, 원격 버전이나 결과는 해당 설정으로 숨기지 않습니다.
+
+설정은 업데이트 캐시와 분리하여
+`$XDG_CONFIG_HOME/session-peer/update-notices.json`에 저장합니다. 환경변수가 없거나
+비어 있으면 `~/.config/session-peer/update-notices.json`을 사용합니다.
+디렉터리와 파일은 안전한 소유권과 비공개 권한이 필요합니다. 안전하지 않은 경로는
+수정하거나 다른 경로로 우회하지 않고 거부합니다. 손상되었거나 향후 스키마를 사용하는
+파일은 알림을 숨기지 않으며 두 명령 모두 덮어쓰지 않으므로 전용 설정 파일을 먼저
+확인하세요. Windows는 아직 버전별 설정을 지원하지 않지만 일반 업데이트·확인과
+전체 알림 비활성화는 계속 사용할 수 있습니다.
+
+안전 검사는 POSIX 소유권과 권한 비트를 확인합니다. ACL(접근 제어 목록)의 전체 검사나
+물리 저장 매체의 내구성 인증은 수행하지 않습니다. 상위 경로의 심볼릭 링크와
+상대 경로로 지정한 설정 위치는 다른 경로로 대체하지 않고 거부합니다. 파일시스템은
+디렉터리 기준 입출력, 파일 잠금, 디렉터리 fsync(동기화)를 지원해야 합니다.
+손상된 설정을 확인하고 백업한 뒤에는 운영자가 해당 전용 JSON 파일만 직접 제거하고
+설정 명령을 다시 실행할 수 있습니다. 쓰기 작업이 실행 중일 수 있다면 잠금 파일을
+제거하지 마세요. 자동 복구는 수행하지 않습니다.
+
패키지 관리자로 설치한 경우 명령어를 설치한 것과 동일한 관리자를 사용하세요:
```bash
@@ -473,8 +507,9 @@ session-peer update --host web-01 # push this program if version
```
원격 업데이트는 최신 GitHub 릴리스가 아니라 **로컬 프로그램의 버전**과 비교합니다.
-원격 버전이 더 최신이더라도 푸시하므로, 배포하기 전에 먼저 확인하고 로컬 프로그램을
-업데이트하세요. 원격 복사본은 GitHub에서 가져오지 않습니다. 원격 버전 검사는
+원격 복사본이 없거나 더 오래된 경우에만 푸시하고, 원격이 더 최신이면 그대로 둡니다.
+배포하기 전에 먼저 확인하고 로컬 프로그램을 업데이트하세요.
+원격 복사본은 GitHub에서 가져오지 않습니다. 원격 버전 검사는
pip/pipx/uv 설치가 아니라 `~/.local/share/session-peer/session_peer.py`만 검사합니다.
원격 업데이트는 해당 독립 실행형 경로와 CLI 링크를 설치합니다. 패키지 관리자로
설치된 원격 CLI의 경우 해당 호스트에서 자체 관리자로 업그레이드하세요.
diff --git a/docs/zh-CN/cli-reference.md b/docs/zh-CN/cli-reference.md
index c0731a9..52d7a6e 100644
--- a/docs/zh-CN/cli-reference.md
+++ b/docs/zh-CN/cli-reference.md
@@ -335,6 +335,19 @@ session-peer device offer-cancel --state /private/device-state --offer-id OFFER-
### 更新
+在 macOS/Linux 上，可以隐藏指定稳定版本的自动运行时通知，或清除此设置：
+
+```bash
+session-peer update --dismiss-runtime-notice 1.3.1
+session-peer update --clear-runtime-notice-dismissal
+```
+
+上述版本仅为示例，并非发行公告。这两个命令只离线修改本地设置，不检查、安装或批准发行版，不能与远程/SSH 选项或检查选项组合使用。只有规范化后完全匹配的版本才会隐藏；后续版本仍可能显示通知，清除后恢复通常行为。显式本地检查仍报告真实的最新版本及是否过时，通知设置作为独立信息提供。远程检查、技能通知和现有的全局关闭通知设置均不变。远程命令仍可能显示调用方 CLI 自身的缓存通知；此时只读取一次本地设置，而不是按目标分别读取。该设置不会隐藏远程版本或结果。
+
+设置独立于更新缓存，保存到 `$XDG_CONFIG_HOME/session-peer/update-notices.json`；环境变量未设置或为空时使用 `~/.config/session-peer/update-notices.json`。目录和文件必须具有安全的所有权和私有权限。不安全的路径会被拒绝，不会自动修复或改用其他路径。无效的设置文件或采用未来版本模式的设置文件不会隐藏任何通知，两个命令也不会覆盖这些文件；请先检查专用设置文件。Windows 暂不支持按版本设置，但仍可使用普通更新、检查和全局关闭通知功能。
+
+安全检查覆盖 POSIX 所有者和权限位，并非完整的 ACL（附加访问权限）审计或物理存储介质持久性认证。祖先路径中的符号链接和显式指定的相对配置路径会被有意拒绝，不会回退。文件系统必须支持基于目录的输入输出、文件锁和目录 fsync（同步）。检查并备份损坏的设置后，操作人员可手动删除该专用 JSON 文件（仅此文件），然后再次运行设置命令；可能仍有写入进程运行时，不要删除锁文件。系统不会自动修复。
+
对于通过包管理器安装的版本，请使用安装该命令的同一包管理器：
```bash
@@ -354,7 +367,7 @@ session-peer update --host web-01 --check # inspect the remote standalon
session-peer update --host web-01 # push this program if versions differ
```
-远程更新是与**本地程序的版本**进行比较，而非最新的 GitHub Release。即使远程版本更新它也会推送；在分发之前，请先检查并更新本地程序。远程副本不会从 GitHub 获取更新。远程版本探测仅检查 `~/.local/share/session-peer/session_peer.py`，不检查 pip/pipx/uv 安装；远程更新会安装该独立路径及其 CLI 链接。对于通过包管理器安装的远程 CLI，请改用该主机上自带的管理器进行升级。
+远程更新是与**本地程序的版本**进行比较，而非最新的 GitHub Release。仅在远程副本不存在或更旧时推送，若远程版本较新则保持不变。在分发之前，请先检查并更新本地程序。远程副本不会从 GitHub 获取更新。远程版本探测仅检查 `~/.local/share/session-peer/session_peer.py`，不检查 pip/pipx/uv 安装；远程更新会安装该独立路径及其 CLI 链接。对于通过包管理器安装的远程 CLI，请改用该主机上自带的管理器进行升级。
无论是本地还是远程的 `update` 都不会刷新 Claude 技能。从所需发布版本的检出目录重新运行 `install.sh` 可同时刷新独立程序和技能。本地的 `session-peer update --check` 和 `session-peer update` 也会填充自动通知所使用的相同缓存。
diff --git a/session_peer.py b/session_peer.py
index 28fd74b..de9c120 100644
--- a/session_peer.py
+++ b/session_peer.py
@@ -6688,6 +6688,39 @@ def update_command() -> str:
return "python -m pip install --upgrade session-peer"
+def runtime_notice_preference_requested(args) -> bool:
+ return (getattr(args, "dismiss_runtime_notice", None) is not None
+ or getattr(args, "clear_runtime_notice_dismissal", False))
+
+
+def runtime_notice_preference_routed(args) -> bool:
+ return bool(getattr(args, "host", None) or getattr(args, "require_ssh_host_key", None)
+ or any(value for key, value in vars(args).items() if key.startswith("ssh_")))
+
+
+def runtime_notice_preference_state() -> dict:
+ # Preference failures never erase an otherwise valid availability notice.
+ try:
+ return read_runtime_notice_preference()
+ except Exception:
+ return {"status": "invalid", "dismissedVersion": None,
+ "reason": "runtime_notice_preference_unavailable"}
+
+
+def runtime_notice_check_metadata(tag, outdated) -> tuple[dict, str]:
+ preference = runtime_notice_preference_state()
+ parsed = stable_version(tag)
+ selected = preference.get("dismissedVersion")
+ dismissed = bool(outdated and parsed is not None and preference.get("status") == "set"
+ and selected == normalized_version(parsed))
+ explanation = ((" Automatic notice for this available release is dismissed."
+ if dismissed else " Stored runtime notice dismissal does not hide this availability result.") +
+ " Selected version: " + selected +
+ ". Clear with: session-peer update --clear-runtime-notice-dismissal."
+ if preference.get("status") == "set" and isinstance(selected, str) else "")
+ return {"runtimeNoticePreference": preference, "runtimeNoticeDismissed": dismissed}, explanation
+
+
def prepare_client_update(args: argparse.Namespace, now: float | None = None,
launcher=None) -> dict | None:
"""Read only local cached state; an expired/missing cache refreshes later."""
@@ -6706,6 +6739,9 @@ def prepare_client_update(args: argparse.Namespace, now: float | None = None,
latest = release_version(state["latest"])
if current is None or latest is None or latest <= current:
return None
+ preference = runtime_notice_preference_state()
+ if preference.get("status") == "set" and preference.get("dismissedVersion") == state["latest"]:
+ return None
checked_at = datetime.fromtimestamp(state["checkedAt"], timezone.utc)
return {
"schemaVersion": UPDATE_CACHE_SCHEMA_VERSION,
@@ -6800,6 +6836,21 @@ def installed_as_distribution() -> bool:
def cmd_update(args: argparse.Namespace) -> int:
+ if runtime_notice_preference_requested(args):
+ if (runtime_notice_preference_routed(args) or getattr(args, "check", False)
+ or (getattr(args, "dismiss_runtime_notice", None) is not None
+ and getattr(args, "clear_runtime_notice_dismissal", False))):
+ raise CcPeerError("Runtime notice preference operations are local and exclusive",
+ {"reason": "runtime_notice_preference_options_conflict", "updated": False})
+ selected = getattr(args, "dismiss_runtime_notice", None)
+ preference = write_runtime_notice_preference(selected)
+ emit(args.json, {"operation": "runtime_notice_preference", "category": "runtime",
+ "updated": False, **preference},
+ ("Runtime notice dismissal cleared; available runtime releases may be shown again."
+ if selected is None else "Runtime notice dismissed for " + preference["dismissedVersion"] +
+ ". This does not mark the release safe/current or install it. Clear with: session-peer update --clear-runtime-notice-dismissal."),
+ command="update")
+ return 0
if not args.host and installed_as_distribution():
command = update_command()
if args.check:
@@ -6811,11 +6862,12 @@ def cmd_update(args: argparse.Namespace) -> int:
latest, current = release_version(tag), release_version(__version__)
outdated = latest is not None and current is not None and current < latest
state = f"{tag} available" if outdated else "up to date"
+ preference, explanation = runtime_notice_check_metadata(tag, outdated)
emit(
args.json,
{"current": __version__, "latest": tag, "outdated": outdated,
- "managedBy": "package-manager", "updateCommand": command},
- human_text(f"session-peer {__version__} — {state}. Upgrade with: {command}"),
+ "managedBy": "package-manager", "updateCommand": command, **preference},
+ human_text(f"session-peer {__version__} — {state}. Upgrade with: {command}" + explanation),
command="update",
)
return 0
@@ -6925,10 +6977,11 @@ def cmd_update(args: argparse.Namespace) -> int:
if args.check:
state = "up to date" if current >= latest else f"{tag} available"
+ preference, explanation = runtime_notice_check_metadata(tag, current < latest)
emit(
args.json,
- {"current": __version__, "latest": tag, "outdated": current < latest},
- human_text(f"session-peer {__version__} — {state}"),
+ {"current": __version__, "latest": tag, "outdated": current < latest, **preference},
+ human_text(f"session-peer {__version__} — {state}" + explanation),
command="update",
)
return 0
@@ -6961,6 +7014,256 @@ def cmd_update(args: argparse.Namespace) -> int:
return 0
+# Local runtime notice preferences; standalone namespace, standard library only.
+RUNTIME_NOTICE_PREFERENCE_BYTES = 4096
+RUNTIME_NOTICE_PREFERENCE_LOCK_SECONDS = 2.0
+
+
+def runtime_notice_cli_version(text):
+ if type(text) is not str or len(text) > 64:
+ raise argparse.ArgumentTypeError("expected an ASCII stable version X.Y.Z")
+ match = re.fullmatch(r"[vV]?(0|[1-9][0-9]*)\.(0|[1-9][0-9]*)\.(0|[1-9][0-9]*)", text, flags=re.ASCII)
+ if match is None:
+ raise argparse.ArgumentTypeError("expected an ASCII stable version X.Y.Z")
+ values = tuple(int(part) for part in match.groups())
+ if any(value > sys.maxsize for value in values):
+ raise argparse.ArgumentTypeError("expected an ASCII stable version X.Y.Z")
+ return ".".join(str(value) for value in values)
+
+
+class RuntimeNoticePreferenceFailure(Exception):
+ def __init__(self, reason):
+ self.reason = reason
+
+
+def _runtime_notice_fail(reason):
+ raise RuntimeNoticePreferenceFailure("runtime_notice_preference_" + reason)
+
+
+def _runtime_notice_supported():
+ return sys.platform in ("linux", "darwin") and fcntl is not None
+
+
+def _runtime_notice_directory_stat(info, *, private=False):
+ owner = os.getuid()
+ if not stat.S_ISDIR(info.st_mode):
+ _runtime_notice_fail("unsafe")
+ if private:
+ # Inherited setgid/sticky bits do not grant group/other access. Keep
+ # owner rwx and no group/other mode bits, without repairing any inode.
+ if info.st_uid != owner or info.st_mode & 0o777 != 0o700:
+ _runtime_notice_fail("unsafe")
+ elif (info.st_uid not in (0, owner) or
+ (info.st_mode & 0o022 and not (info.st_uid == 0 and info.st_mode & stat.S_ISVTX))):
+ _runtime_notice_fail("unsafe")
+
+
+def _runtime_notice_file_stat(info):
+ if (not stat.S_ISREG(info.st_mode) or info.st_uid != os.getuid()
+ or info.st_mode & 0o077 or info.st_nlink != 1):
+ _runtime_notice_fail("unsafe")
+
+
+@contextlib.contextmanager
+def _runtime_notice_directory(*, create=False):
+ configured = os.environ.get("XDG_CONFIG_HOME")
+ base = configured if configured else str(Path.home() / ".config")
+ if (type(base) is not str or not base.startswith("/") or "\x00" in base
+ or any(part in (".", "..") for part in base.split("/"))):
+ _runtime_notice_fail("unsafe")
+ components = [part for part in base.split("/") if part] + ["session-peer"]
+ flags = os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | getattr(os, "O_CLOEXEC", 0)
+ fd = os.open("/", flags)
+ try:
+ _runtime_notice_directory_stat(os.fstat(fd))
+ for index, name in enumerate(components):
+ try:
+ child = os.open(name, flags, dir_fd=fd)
+ except FileNotFoundError:
+ if not create:
+ yield None
+ return
+ try:
+ os.mkdir(name, mode=0o700, dir_fd=fd)
+ os.fsync(fd)
+ except FileExistsError:
+ pass
+ child = os.open(name, flags, dir_fd=fd)
+ except OSError as exc:
+ if exc.errno in (errno.ELOOP, errno.ENOTDIR):
+ _runtime_notice_fail("unsafe")
+ raise
+ try:
+ _runtime_notice_directory_stat(os.fstat(child), private=index == len(components)-1)
+ except BaseException:
+ os.close(child)
+ raise
+ os.close(fd)
+ fd = child
+ yield fd
+ finally:
+ os.close(fd)
+
+
+def _runtime_notice_open_file(directory, name, *, create=False):
+ try:
+ before = os.stat(name, dir_fd=directory, follow_symlinks=False)
+ except FileNotFoundError:
+ if not create:
+ return None
+ before = None
+ if before is not None:
+ _runtime_notice_file_stat(before)
+ if not create and before.st_size > RUNTIME_NOTICE_PREFERENCE_BYTES:
+ _runtime_notice_fail("invalid")
+ flags = (os.O_RDWR if create else os.O_RDONLY) | os.O_NOFOLLOW | os.O_NONBLOCK | getattr(os, "O_CLOEXEC", 0)
+ if create and before is None:
+ flags |= os.O_CREAT | os.O_EXCL
+ try:
+ fd = os.open(name, flags, 0o600, dir_fd=directory)
+ except FileExistsError:
+ if not create or before is not None:
+ raise
+ # Another cooperative first writer may have just created the stable
+ # lock. Validate and bind its existing inode instead of replacing it.
+ before = os.stat(name, dir_fd=directory, follow_symlinks=False)
+ _runtime_notice_file_stat(before)
+ fd = os.open(name, flags & ~(os.O_CREAT | os.O_EXCL), 0o600, dir_fd=directory)
+ try:
+ after = os.fstat(fd)
+ _runtime_notice_file_stat(after)
+ if before is not None and (before.st_dev, before.st_ino) != (after.st_dev, after.st_ino):
+ _runtime_notice_fail("unsafe")
+ if not create and after.st_size > RUNTIME_NOTICE_PREFERENCE_BYTES:
+ _runtime_notice_fail("invalid")
+ return fd
+ except BaseException:
+ os.close(fd)
+ raise
+
+
+def _runtime_notice_decode(raw):
+ def pairs(items):
+ result = {}
+ for key, value in items:
+ if key in result:
+ _runtime_notice_fail("invalid")
+ result[key] = value
+ return result
+ def constant(value):
+ _runtime_notice_fail("invalid")
+ try:
+ value = json.loads(raw.decode("utf-8"), object_pairs_hook=pairs, parse_constant=constant)
+ if (type(value) is not dict or set(value) != {"schemaVersion", "runtime"}
+ or type(value["schemaVersion"]) is not int or value["schemaVersion"] != 1
+ or type(value["runtime"]) is not dict or set(value["runtime"]) != {"dismissedVersion"}):
+ _runtime_notice_fail("invalid")
+ version = value["runtime"]["dismissedVersion"]
+ if version is not None and runtime_notice_cli_version(version) != version:
+ _runtime_notice_fail("invalid")
+ return version
+ except (ValueError, TypeError, UnicodeError, argparse.ArgumentTypeError, RecursionError):
+ _runtime_notice_fail("invalid")
+
+
+def _runtime_notice_read(directory):
+ fd = _runtime_notice_open_file(directory, "update-notices.json")
+ if fd is None:
+ return {"status": "missing", "dismissedVersion": None}
+ try:
+ chunks, size = [], 0
+ while size <= RUNTIME_NOTICE_PREFERENCE_BYTES:
+ chunk = os.read(fd, RUNTIME_NOTICE_PREFERENCE_BYTES + 1 - size)
+ if not chunk:
+ break
+ chunks.append(chunk)
+ size += len(chunk)
+ raw = b"".join(chunks)
+ if len(raw) > RUNTIME_NOTICE_PREFERENCE_BYTES:
+ _runtime_notice_fail("invalid")
+ version = _runtime_notice_decode(raw)
+ return {"status": "clear" if version is None else "set", "dismissedVersion": version}
+ finally:
+ os.close(fd)
+
+
+def read_runtime_notice_preference():
+ if not _runtime_notice_supported():
+ return {"status": "unsupported", "dismissedVersion": None, "reason": "runtime_notice_preference_unsupported"}
+ try:
+ with _runtime_notice_directory() as directory:
+ return ({"status": "missing", "dismissedVersion": None} if directory is None
+ else _runtime_notice_read(directory))
+ except Exception as exc:
+ reason = exc.reason if isinstance(exc, RuntimeNoticePreferenceFailure) else "runtime_notice_preference_unavailable"
+ return {"status": "invalid", "dismissedVersion": None, "reason": reason}
+
+
+def write_runtime_notice_preference(version_or_none):
+ if not _runtime_notice_supported():
+ raise CcPeerError("Runtime notice preference refused", {"reason": "runtime_notice_preference_unsupported", "updated": False})
+ replaced = False
+ try:
+ try:
+ version = None if version_or_none is None else runtime_notice_cli_version(version_or_none)
+ except argparse.ArgumentTypeError:
+ _runtime_notice_fail("invalid")
+ with _runtime_notice_directory(create=True) as directory:
+ lock = _runtime_notice_open_file(directory, "update-notices.lock", create=True)
+ temporary, identity = None, None
+ try:
+ deadline = time.monotonic() + RUNTIME_NOTICE_PREFERENCE_LOCK_SECONDS
+ while True:
+ try:
+ fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
+ break
+ except BlockingIOError:
+ remaining = deadline - time.monotonic()
+ if remaining <= 0:
+ _runtime_notice_fail("busy")
+ time.sleep(min(0.01, remaining))
+ current = _runtime_notice_read(directory)
+ status = "clear" if version is None else "set"
+ if current["status"] != "missing" and current["dismissedVersion"] == version:
+ return {"status": status, "dismissedVersion": version, "changed": False}
+ raw = json.dumps({"schemaVersion": 1, "runtime": {"dismissedVersion": version}},
+ separators=(",", ":"), allow_nan=False).encode("utf-8")
+ temporary = ".update-notices-" + uuid.uuid4().hex + ".tmp"
+ fd = os.open(temporary, os.O_CREAT | os.O_EXCL | os.O_WRONLY | os.O_NOFOLLOW,
+ 0o600, dir_fd=directory)
+ try:
+ info = os.fstat(fd)
+ identity = (info.st_dev, info.st_ino)
+ _runtime_notice_file_stat(info)
+ offset = 0
+ while offset < len(raw):
+ count = os.write(fd, raw[offset:])
+ if count <= 0:
+ raise OSError()
+ offset += count
+ os.fsync(fd)
+ finally:
+ os.close(fd)
+ os.replace(temporary, "update-notices.json", src_dir_fd=directory, dst_dir_fd=directory)
+ replaced, temporary = True, None
+ os.fsync(directory)
+ return {"status": status, "dismissedVersion": version, "changed": True}
+ finally:
+ try:
+ if temporary is not None and identity is not None:
+ info = os.stat(temporary, dir_fd=directory, follow_symlinks=False)
+ if (info.st_dev, info.st_ino) == identity and info.st_uid == os.getuid():
+ os.unlink(temporary, dir_fd=directory)
+ finally:
+ os.close(lock)
+ except Exception as exc:
+ reason = exc.reason if isinstance(exc, RuntimeNoticePreferenceFailure) else "runtime_notice_preference_write_failed"
+ details = {"reason": reason, "updated": False}
+ if replaced:
+ details["preferenceMayHaveChanged"] = True
+ raise CcPeerError("Runtime notice preference refused", details) from None
+
+
"""Bounded opt-in POSIX message input; ordinary CLI input is unchanged."""
import codecs
@@ -10513,9 +10816,14 @@ def add_common(sub: argparse.ArgumentParser) -> None:
updating = subparsers.add_parser("update", help="update this installation")
add_common(updating)
- updating.add_argument(
+ update_action = updating.add_mutually_exclusive_group()
+ update_action.add_argument(
"--check", action="store_true", help="report the available version, change nothing"
)
+ update_action.add_argument("--dismiss-runtime-notice", type=runtime_notice_cli_version,
+ metavar="VERSION", help="dismiss only this runtime release's ambient notice; no installation")
+ update_action.add_argument("--clear-runtime-notice-dismissal", action="store_true",
+ help="clear the local runtime notice dismissal; no installation")
updating.set_defaults(func=cmd_update)
# Deliberately not add_common: setup must not gain executable config input.
@@ -10646,6 +10954,10 @@ def _main(argv: list[str] | None = None) -> int:
if args.json and output_format == "text":
parser.error("--json conflicts with --output-format text")
args.json = args.json or output_format == "json"
+ if runtime_notice_preference_requested(args):
+ if runtime_notice_preference_routed(args):
+ parser.error("Runtime notice preferences are local only; do not use host or SSH options")
+ args.no_update_notice = True
try:
prepare_trusted_ssh_config(parser, args)
except CcPeerError as exc:
diff --git a/session_peer_core/cli.py b/session_peer_core/cli.py
index cf9cfa0..ffa8c4e 100644
--- a/session_peer_core/cli.py
+++ b/session_peer_core/cli.py
@@ -208,9 +208,14 @@ def add_common(sub: argparse.ArgumentParser) -> None:
updating = subparsers.add_parser("update", help="update this installation")
add_common(updating)
- updating.add_argument(
+ update_action = updating.add_mutually_exclusive_group()
+ update_action.add_argument(
"--check", action="store_true", help="report the available version, change nothing"
)
+ update_action.add_argument("--dismiss-runtime-notice", type=runtime_notice_cli_version,
+ metavar="VERSION", help="dismiss only this runtime release's ambient notice; no installation")
+ update_action.add_argument("--clear-runtime-notice-dismissal", action="store_true",
+ help="clear the local runtime notice dismissal; no installation")
updating.set_defaults(func=cmd_update)
# Deliberately not add_common: setup must not gain executable config input.
@@ -341,6 +346,10 @@ def _main(argv: list[str] | None = None) -> int:
if args.json and output_format == "text":
parser.error("--json conflicts with --output-format text")
args.json = args.json or output_format == "json"
+ if runtime_notice_preference_requested(args):
+ if runtime_notice_preference_routed(args):
+ parser.error("Runtime notice preferences are local only; do not use host or SSH options")
+ args.no_update_notice = True
try:
prepare_trusted_ssh_config(parser, args)
except CcPeerError as exc:
diff --git a/session_peer_core/update_preferences.py b/session_peer_core/update_preferences.py
new file mode 100644
index 0000000..3e3bd79
--- /dev/null
+++ b/session_peer_core/update_preferences.py
@@ -0,0 +1,248 @@
+# Local runtime notice preferences; standalone namespace, standard library only.
+RUNTIME_NOTICE_PREFERENCE_BYTES = 4096
+RUNTIME_NOTICE_PREFERENCE_LOCK_SECONDS = 2.0
+
+
+def runtime_notice_cli_version(text):
+ if type(text) is not str or len(text) > 64:
+ raise argparse.ArgumentTypeError("expected an ASCII stable version X.Y.Z")
+ match = re.fullmatch(r"[vV]?(0|[1-9][0-9]*)\.(0|[1-9][0-9]*)\.(0|[1-9][0-9]*)", text, flags=re.ASCII)
+ if match is None:
+ raise argparse.ArgumentTypeError("expected an ASCII stable version X.Y.Z")
+ values = tuple(int(part) for part in match.groups())
+ if any(value > sys.maxsize for value in values):
+ raise argparse.ArgumentTypeError("expected an ASCII stable version X.Y.Z")
+ return ".".join(str(value) for value in values)
+
+
+class RuntimeNoticePreferenceFailure(Exception):
+ def __init__(self, reason):
+ self.reason = reason
+
+
+def _runtime_notice_fail(reason):
+ raise RuntimeNoticePreferenceFailure("runtime_notice_preference_" + reason)
+
+
+def _runtime_notice_supported():
+ return sys.platform in ("linux", "darwin") and fcntl is not None
+
+
+def _runtime_notice_directory_stat(info, *, private=False):
+ owner = os.getuid()
+ if not stat.S_ISDIR(info.st_mode):
+ _runtime_notice_fail("unsafe")
+ if private:
+ # Inherited setgid/sticky bits do not grant group/other access. Keep
+ # owner rwx and no group/other mode bits, without repairing any inode.
+ if info.st_uid != owner or info.st_mode & 0o777 != 0o700:
+ _runtime_notice_fail("unsafe")
+ elif (info.st_uid not in (0, owner) or
+ (info.st_mode & 0o022 and not (info.st_uid == 0 and info.st_mode & stat.S_ISVTX))):
+ _runtime_notice_fail("unsafe")
+
+
+def _runtime_notice_file_stat(info):
+ if (not stat.S_ISREG(info.st_mode) or info.st_uid != os.getuid()
+ or info.st_mode & 0o077 or info.st_nlink != 1):
+ _runtime_notice_fail("unsafe")
+
+
+@contextlib.contextmanager
+def _runtime_notice_directory(*, create=False):
+ configured = os.environ.get("XDG_CONFIG_HOME")
+ base = configured if configured else str(Path.home() / ".config")
+ if (type(base) is not str or not base.startswith("/") or "\x00" in base
+ or any(part in (".", "..") for part in base.split("/"))):
+ _runtime_notice_fail("unsafe")
+ components = [part for part in base.split("/") if part] + ["session-peer"]
+ flags = os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | getattr(os, "O_CLOEXEC", 0)
+ fd = os.open("/", flags)
+ try:
+ _runtime_notice_directory_stat(os.fstat(fd))
+ for index, name in enumerate(components):
+ try:
+ child = os.open(name, flags, dir_fd=fd)
+ except FileNotFoundError:
+ if not create:
+ yield None
+ return
+ try:
+ os.mkdir(name, mode=0o700, dir_fd=fd)
+ os.fsync(fd)
+ except FileExistsError:
+ pass
+ child = os.open(name, flags, dir_fd=fd)
+ except OSError as exc:
+ if exc.errno in (errno.ELOOP, errno.ENOTDIR):
+ _runtime_notice_fail("unsafe")
+ raise
+ try:
+ _runtime_notice_directory_stat(os.fstat(child), private=index == len(components)-1)
+ except BaseException:
+ os.close(child)
+ raise
+ os.close(fd)
+ fd = child
+ yield fd
+ finally:
+ os.close(fd)
+
+
+def _runtime_notice_open_file(directory, name, *, create=False):
+ try:
+ before = os.stat(name, dir_fd=directory, follow_symlinks=False)
+ except FileNotFoundError:
+ if not create:
+ return None
+ before = None
+ if before is not None:
+ _runtime_notice_file_stat(before)
+ if not create and before.st_size > RUNTIME_NOTICE_PREFERENCE_BYTES:
+ _runtime_notice_fail("invalid")
+ flags = (os.O_RDWR if create else os.O_RDONLY) | os.O_NOFOLLOW | os.O_NONBLOCK | getattr(os, "O_CLOEXEC", 0)
+ if create and before is None:
+ flags |= os.O_CREAT | os.O_EXCL
+ try:
+ fd = os.open(name, flags, 0o600, dir_fd=directory)
+ except FileExistsError:
+ if not create or before is not None:
+ raise
+ # Another cooperative first writer may have just created the stable
+ # lock. Validate and bind its existing inode instead of replacing it.
+ before = os.stat(name, dir_fd=directory, follow_symlinks=False)
+ _runtime_notice_file_stat(before)
+ fd = os.open(name, flags & ~(os.O_CREAT | os.O_EXCL), 0o600, dir_fd=directory)
+ try:
+ after = os.fstat(fd)
+ _runtime_notice_file_stat(after)
+ if before is not None and (before.st_dev, before.st_ino) != (after.st_dev, after.st_ino):
+ _runtime_notice_fail("unsafe")
+ if not create and after.st_size > RUNTIME_NOTICE_PREFERENCE_BYTES:
+ _runtime_notice_fail("invalid")
+ return fd
+ except BaseException:
+ os.close(fd)
+ raise
+
+
+def _runtime_notice_decode(raw):
+ def pairs(items):
+ result = {}
+ for key, value in items:
+ if key in result:
+ _runtime_notice_fail("invalid")
+ result[key] = value
+ return result
+ def constant(value):
+ _runtime_notice_fail("invalid")
+ try:
+ value = json.loads(raw.decode("utf-8"), object_pairs_hook=pairs, parse_constant=constant)
+ if (type(value) is not dict or set(value) != {"schemaVersion", "runtime"}
+ or type(value["schemaVersion"]) is not int or value["schemaVersion"] != 1
+ or type(value["runtime"]) is not dict or set(value["runtime"]) != {"dismissedVersion"}):
+ _runtime_notice_fail("invalid")
+ version = value["runtime"]["dismissedVersion"]
+ if version is not None and runtime_notice_cli_version(version) != version:
+ _runtime_notice_fail("invalid")
+ return version
+ except (ValueError, TypeError, UnicodeError, argparse.ArgumentTypeError, RecursionError):
+ _runtime_notice_fail("invalid")
+
+
+def _runtime_notice_read(directory):
+ fd = _runtime_notice_open_file(directory, "update-notices.json")
+ if fd is None:
+ return {"status": "missing", "dismissedVersion": None}
+ try:
+ chunks, size = [], 0
+ while size <= RUNTIME_NOTICE_PREFERENCE_BYTES:
+ chunk = os.read(fd, RUNTIME_NOTICE_PREFERENCE_BYTES + 1 - size)
+ if not chunk:
+ break
+ chunks.append(chunk)
+ size += len(chunk)
+ raw = b"".join(chunks)
+ if len(raw) > RUNTIME_NOTICE_PREFERENCE_BYTES:
+ _runtime_notice_fail("invalid")
+ version = _runtime_notice_decode(raw)
+ return {"status": "clear" if version is None else "set", "dismissedVersion": version}
+ finally:
+ os.close(fd)
+
+
+def read_runtime_notice_preference():
+ if not _runtime_notice_supported():
+ return {"status": "unsupported", "dismissedVersion": None, "reason": "runtime_notice_preference_unsupported"}
+ try:
+ with _runtime_notice_directory() as directory:
+ return ({"status": "missing", "dismissedVersion": None} if directory is None
+ else _runtime_notice_read(directory))
+ except Exception as exc:
+ reason = exc.reason if isinstance(exc, RuntimeNoticePreferenceFailure) else "runtime_notice_preference_unavailable"
+ return {"status": "invalid", "dismissedVersion": None, "reason": reason}
+
+
+def write_runtime_notice_preference(version_or_none):
+ if not _runtime_notice_supported():
+ raise CcPeerError("Runtime notice preference refused", {"reason": "runtime_notice_preference_unsupported", "updated": False})
+ replaced = False
+ try:
+ try:
+ version = None if version_or_none is None else runtime_notice_cli_version(version_or_none)
+ except argparse.ArgumentTypeError:
+ _runtime_notice_fail("invalid")
+ with _runtime_notice_directory(create=True) as directory:
+ lock = _runtime_notice_open_file(directory, "update-notices.lock", create=True)
+ temporary, identity = None, None
+ try:
+ deadline = time.monotonic() + RUNTIME_NOTICE_PREFERENCE_LOCK_SECONDS
+ while True:
+ try:
+ fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
+ break
+ except BlockingIOError:
+ remaining = deadline - time.monotonic()
+ if remaining <= 0:
+ _runtime_notice_fail("busy")
+ time.sleep(min(0.01, remaining))
+ current = _runtime_notice_read(directory)
+ status = "clear" if version is None else "set"
+ if current["status"] != "missing" and current["dismissedVersion"] == version:
+ return {"status": status, "dismissedVersion": version, "changed": False}
+ raw = json.dumps({"schemaVersion": 1, "runtime": {"dismissedVersion": version}},
+ separators=(",", ":"), allow_nan=False).encode("utf-8")
+ temporary = ".update-notices-" + uuid.uuid4().hex + ".tmp"
+ fd = os.open(temporary, os.O_CREAT | os.O_EXCL | os.O_WRONLY | os.O_NOFOLLOW,
+ 0o600, dir_fd=directory)
+ try:
+ info = os.fstat(fd)
+ identity = (info.st_dev, info.st_ino)
+ _runtime_notice_file_stat(info)
+ offset = 0
+ while offset < len(raw):
+ count = os.write(fd, raw[offset:])
+ if count <= 0:
+ raise OSError()
+ offset += count
+ os.fsync(fd)
+ finally:
+ os.close(fd)
+ os.replace(temporary, "update-notices.json", src_dir_fd=directory, dst_dir_fd=directory)
+ replaced, temporary = True, None
+ os.fsync(directory)
+ return {"status": status, "dismissedVersion": version, "changed": True}
+ finally:
+ try:
+ if temporary is not None and identity is not None:
+ info = os.stat(temporary, dir_fd=directory, follow_symlinks=False)
+ if (info.st_dev, info.st_ino) == identity and info.st_uid == os.getuid():
+ os.unlink(temporary, dir_fd=directory)
+ finally:
+ os.close(lock)
+ except Exception as exc:
+ reason = exc.reason if isinstance(exc, RuntimeNoticePreferenceFailure) else "runtime_notice_preference_write_failed"
+ details = {"reason": reason, "updated": False}
+ if replaced:
+ details["preferenceMayHaveChanged"] = True
+ raise CcPeerError("Runtime notice preference refused", details) from None
diff --git a/session_peer_core/updates.py b/session_peer_core/updates.py
index 23c0d01..10bb935 100644
--- a/session_peer_core/updates.py
+++ b/session_peer_core/updates.py
@@ -373,6 +373,39 @@ def update_command() -> str:
return "python -m pip install --upgrade session-peer"
+def runtime_notice_preference_requested(args) -> bool:
+ return (getattr(args, "dismiss_runtime_notice", None) is not None
+ or getattr(args, "clear_runtime_notice_dismissal", False))
+
+
+def runtime_notice_preference_routed(args) -> bool:
+ return bool(getattr(args, "host", None) or getattr(args, "require_ssh_host_key", None)
+ or any(value for key, value in vars(args).items() if key.startswith("ssh_")))
+
+
+def runtime_notice_preference_state() -> dict:
+ # Preference failures never erase an otherwise valid availability notice.
+ try:
+ return read_runtime_notice_preference()
+ except Exception:
+ return {"status": "invalid", "dismissedVersion": None,
+ "reason": "runtime_notice_preference_unavailable"}
+
+
+def runtime_notice_check_metadata(tag, outdated) -> tuple[dict, str]:
+ preference = runtime_notice_preference_state()
+ parsed = stable_version(tag)
+ selected = preference.get("dismissedVersion")
+ dismissed = bool(outdated and parsed is not None and preference.get("status") == "set"
+ and selected == normalized_version(parsed))
+ explanation = ((" Automatic notice for this available release is dismissed."
+ if dismissed else " Stored runtime notice dismissal does not hide this availability result.") +
+ " Selected version: " + selected +
+ ". Clear with: session-peer update --clear-runtime-notice-dismissal."
+ if preference.get("status") == "set" and isinstance(selected, str) else "")
+ return {"runtimeNoticePreference": preference, "runtimeNoticeDismissed": dismissed}, explanation
+
+
def prepare_client_update(args: argparse.Namespace, now: float | None = None,
launcher=None) -> dict | None:
"""Read only local cached state; an expired/missing cache refreshes later."""
@@ -391,6 +424,9 @@ def prepare_client_update(args: argparse.Namespace, now: float | None = None,
latest = release_version(state["latest"])
if current is None or latest is None or latest <= current:
return None
+ preference = runtime_notice_preference_state()
+ if preference.get("status") == "set" and preference.get("dismissedVersion") == state["latest"]:
+ return None
checked_at = datetime.fromtimestamp(state["checkedAt"], timezone.utc)
return {
"schemaVersion": UPDATE_CACHE_SCHEMA_VERSION,
@@ -485,6 +521,21 @@ def installed_as_distribution() -> bool:
def cmd_update(args: argparse.Namespace) -> int:
+ if runtime_notice_preference_requested(args):
+ if (runtime_notice_preference_routed(args) or getattr(args, "check", False)
+ or (getattr(args, "dismiss_runtime_notice", None) is not None
+ and getattr(args, "clear_runtime_notice_dismissal", False))):
+ raise CcPeerError("Runtime notice preference operations are local and exclusive",
+ {"reason": "runtime_notice_preference_options_conflict", "updated": False})
+ selected = getattr(args, "dismiss_runtime_notice", None)
+ preference = write_runtime_notice_preference(selected)
+ emit(args.json, {"operation": "runtime_notice_preference", "category": "runtime",
+ "updated": False, **preference},
+ ("Runtime notice dismissal cleared; available runtime releases may be shown again."
+ if selected is None else "Runtime notice dismissed for " + preference["dismissedVersion"] +
+ ". This does not mark the release safe/current or install it. Clear with: session-peer update --clear-runtime-notice-dismissal."),
+ command="update")
+ return 0
if not args.host and installed_as_distribution():
command = update_command()
if args.check:
@@ -496,11 +547,12 @@ def cmd_update(args: argparse.Namespace) -> int:
latest, current = release_version(tag), release_version(__version__)
outdated = latest is not None and current is not None and current < latest
state = f"{tag} available" if outdated else "up to date"
+ preference, explanation = runtime_notice_check_metadata(tag, outdated)
emit(
args.json,
{"current": __version__, "latest": tag, "outdated": outdated,
- "managedBy": "package-manager", "updateCommand": command},
- human_text(f"session-peer {__version__} — {state}. Upgrade with: {command}"),
+ "managedBy": "package-manager", "updateCommand": command, **preference},
+ human_text(f"session-peer {__version__} — {state}. Upgrade with: {command}" + explanation),
command="update",
)
return 0
@@ -610,10 +662,11 @@ def cmd_update(args: argparse.Namespace) -> int:
if args.check:
state = "up to date" if current >= latest else f"{tag} available"
+ preference, explanation = runtime_notice_check_metadata(tag, current < latest)
emit(
args.json,
- {"current": __version__, "latest": tag, "outdated": current < latest},
- human_text(f"session-peer {__version__} — {state}"),
+ {"current": __version__, "latest": tag, "outdated": current < latest, **preference},
+ human_text(f"session-peer {__version__} — {state}" + explanation),
command="update",
)
return 0
diff --git a/tests/core/test_update_notice_dismissal.py b/tests/core/test_update_notice_dismissal.py
new file mode 100644
index 0000000..3c9efb3
--- /dev/null
+++ b/tests/core/test_update_notice_dismissal.py
@@ -0,0 +1,282 @@
+"""Offline runtime preference integration; no user state, network or installers."""
+from __future__ import annotations
+
+import argparse
+import contextlib
+import io
+import json
+import os
+import unittest
+from unittest import mock
+
+import session_peer
+
+
+class RuntimeNoticeDismissal(unittest.TestCase):
+ def setUp(self):
+ self.peer = session_peer # Missing generated code must fail, not fall back.
+ self.stack = contextlib.ExitStack()
+ self.addCleanup(self.stack.close)
+ self.stack.enter_context(mock.patch.object(self.peer, '__version__', '1.0.0'))
+ self.stack.enter_context(mock.patch.object(self.peer, '_CLIENT_UPDATE_NOTICE', None))
+ self.stack.enter_context(mock.patch.object(self.peer, '_SKILL_UPDATE_NOTICES', []))
+ self.read = self.stack.enter_context(mock.patch.object(self.peer, 'read_runtime_notice_preference',
+ return_value={'status': 'clear', 'dismissedVersion': None}))
+ self.write = self.stack.enter_context(mock.patch.object(self.peer, 'write_runtime_notice_preference',
+ return_value={'status': 'set', 'dismissedVersion': '2.0.0', 'changed': True}))
+ self.latest = self.stack.enter_context(mock.patch.object(self.peer, 'latest_release',
+ side_effect=AssertionError('unexpected network')))
+ self.cache_write = self.stack.enter_context(mock.patch.object(self.peer, 'write_update_cache'))
+ self.distribution = self.stack.enter_context(mock.patch.object(self.peer, 'installed_as_distribution', return_value=False))
+ self.stack.enter_context(mock.patch.object(self.peer, 'installed_skills', return_value=[]))
+ self.stack.enter_context(mock.patch.dict(os.environ, {self.peer.UPDATE_NOTICE_ENV: ''}))
+
+ def args(self, **changes):
+ return argparse.Namespace(command='send', no_update_notice=False, host=[], **changes)
+
+ def cache(self, latest='2.0.0'):
+ return mock.patch.object(self.peer, 'read_update_cache', return_value={
+ 'status': 'fresh', 'latest': latest, 'checkedAt': 1000})
+
+ def main(self, argv, *, ambient=False):
+ output = io.StringIO()
+ with contextlib.redirect_stdout(output):
+ if ambient:
+ with mock.patch.object(self.peer.sys, 'argv', ['session-peer', *argv]):
+ result = self.peer.main()
+ else:
+ result = self.peer.main(argv)
+ return result, output.getvalue()
+
+ def test_exact_dismissal_later_release_and_clear_are_runtime_only(self):
+ self.read.return_value = {'status': 'set', 'dismissedVersion': '2.0.0'}
+ with self.cache():
+ self.assertIsNone(self.peer.prepare_client_update(self.args(), now=1001))
+ with self.cache('2.0.1'):
+ self.assertEqual(self.peer.prepare_client_update(self.args(), now=1001)['latest'], '2.0.1')
+ self.read.return_value = {'status': 'clear', 'dismissedVersion': None}
+ with self.cache():
+ self.assertEqual(self.peer.prepare_client_update(self.args(), now=1001)['latest'], '2.0.0')
+ self.assertEqual(self.read.call_count, 3)
+ self.latest.assert_not_called()
+ self.cache_write.assert_not_called()
+
+ def test_invalid_preference_or_read_failure_does_not_hide_availability(self):
+ for preference in ({'status': 'invalid', 'dismissedVersion': None},
+ {'status': 'unsupported', 'dismissedVersion': None}):
+ self.read.return_value = preference
+ with self.cache():
+ self.assertEqual(self.peer.prepare_client_update(self.args(), now=1001)['status'], 'available')
+ self.read.side_effect = OSError('fixture unreadable')
+ with self.cache():
+ self.assertEqual(self.peer.prepare_client_update(self.args(), now=1001)['status'], 'available')
+
+ def test_global_opt_out_and_irrelevant_candidate_do_not_read_preference(self):
+ with self.cache(), mock.patch.dict(os.environ, {self.peer.UPDATE_NOTICE_ENV: 'true'}):
+ self.assertIsNone(self.peer.prepare_client_update(self.args(), now=1001))
+ with self.cache():
+ args = self.args()
+ args.no_update_notice = True
+ self.assertIsNone(self.peer.prepare_client_update(args, now=1001))
+ with self.cache('1.0.0'):
+ self.assertIsNone(self.peer.prepare_client_update(self.args(), now=1001))
+ self.read.assert_not_called()
+
+ def test_stale_cache_preserves_existing_refresh_without_preference_read(self):
+ launch = mock.Mock()
+ with mock.patch.object(self.peer, 'read_update_cache', return_value={'status': 'expired'}):
+ self.assertIsNone(self.peer.prepare_client_update(self.args(), launcher=launch))
+ launch.assert_called_once_with()
+ self.read.assert_not_called()
+ self.latest.assert_not_called()
+
+ def test_preference_operations_are_offline_and_precede_package_ownership(self):
+ for option, selected, state in (
+ (['--dismiss-runtime-notice', 'v2.0.0'], '2.0.0', 'set'),
+ (['--clear-runtime-notice-dismissal'], None, 'clear')):
+ self.write.return_value = {'status': state, 'dismissedVersion': selected, 'changed': True}
+ with mock.patch.object(self.peer.subprocess, 'Popen', side_effect=AssertionError('unexpected launch')), \
+ mock.patch.object(self.peer, 'push_to_remote', side_effect=AssertionError('unexpected SSH')), \
+ mock.patch.object(self.peer, 'verified_release_download', side_effect=AssertionError('unexpected install')):
+ code, output = self.main(['update', *option, '--json'], ambient=True)
+ result = json.loads(output)
+ self.assertEqual(code, 0)
+ self.assertEqual(result['operation'], 'runtime_notice_preference')
+ self.assertEqual(result['category'], 'runtime')
+ self.assertFalse(result['updated'])
+ self.assertEqual(result['dismissedVersion'], selected)
+ self.assertNotIn('latest', result)
+ self.write.assert_called_with(selected)
+ self.distribution.assert_not_called()
+ self.latest.assert_not_called()
+ self.read.assert_not_called()
+ self.cache_write.assert_not_called()
+
+ def test_conflicting_routes_or_versions_reject_before_advisory_preparation(self):
+ invalid = [
+ ['--dismiss-runtime-notice', '2.0.0', '--check'],
+ ['--dismiss-runtime-notice', '2.0.0', '--clear-runtime-notice-dismissal'],
+ ['--dismiss-runtime-notice', '2.0.0', '--host', 'elsewhere'],
+ ['--clear-runtime-notice-dismissal', '--ssh-opt=-p'],
+ ['--clear-runtime-notice-dismissal', '--ssh-identity'],
+ ['--clear-runtime-notice-dismissal', '--require-ssh-host-key', 'SHA256:' + 'a' * 43],
+ ]
+ invalid += [['--dismiss-runtime-notice', value] for value in ('1.2', '1.2.3rc1', '01.2.3', '１.2.3')]
+ for options in invalid:
+ with self.subTest(options=options), \
+ mock.patch.object(self.peer, 'prepare_client_update') as ambient, \
+ mock.patch.object(self.peer, 'prepare_skill_updates') as skill, \
+ contextlib.redirect_stderr(io.StringIO()), self.assertRaises(SystemExit):
+ self.main(['update', *options], ambient=True)
+ ambient.assert_not_called()
+ skill.assert_not_called()
+ self.write.assert_not_called()
+ self.latest.assert_not_called()
+
+ def test_future_ssh_configuration_fields_are_rejected_before_advisory(self):
+ parsed = self.peer.build_parser().parse_args(['update', '--clear-runtime-notice-dismissal'])
+ parsed.ssh_config_file = '/fixture/config'
+ with mock.patch.object(self.peer, 'build_parser') as parser, \
+ mock.patch.object(self.peer, 'prepare_client_update') as ambient, \
+ mock.patch.object(self.peer, 'prepare_skill_updates') as skill:
+ parser.return_value.parse_args.return_value = parsed
+ parser.return_value.error.side_effect = SystemExit(2)
+ with self.assertRaises(SystemExit):
+ self.main(['update', '--clear-runtime-notice-dismissal'], ambient=True)
+ ambient.assert_not_called()
+ skill.assert_not_called()
+ self.write.assert_not_called()
+
+ def test_explicit_checks_remain_truthful_for_managed_and_standalone(self):
+ self.read.return_value = {'status': 'set', 'dismissedVersion': '2.0.0'}
+ self.latest.side_effect = None
+ self.latest.return_value = ('v2.0.0', 'offline-fixture-url')
+ for managed in (False, True):
+ self.distribution.return_value = managed
+ code, output = self.main(['update', '--check', '--json', '--no-update-notice'])
+ result = json.loads(output)
+ self.assertEqual(code, 0)
+ self.assertEqual(result['latest'], 'v2.0.0')
+ self.assertTrue(result['outdated'])
+ self.assertTrue(result['runtimeNoticeDismissed'])
+ self.assertEqual(result['runtimeNoticePreference']['dismissedVersion'], '2.0.0')
+ if managed:
+ self.assertEqual(result['managedBy'], 'package-manager')
+ code, human = self.main(['update', '--check', '--no-update-notice'])
+ self.assertEqual(code, 0)
+ self.assertIn('v2.0.0 available', human)
+ self.assertIn('Automatic notice for this available release is dismissed.', human)
+ self.assertIn('--clear-runtime-notice-dismissal', human)
+
+ def test_human_check_distinguishes_nonmatching_or_current_dismissal(self):
+ for latest, selected in (('2.0.0', '1.5.0'), ('1.0.0', '1.0.0')):
+ self.read.return_value = {'status': 'set', 'dismissedVersion': selected}
+ with mock.patch.object(self.peer, 'latest_release', return_value=(latest, 'offline-fixture-url')):
+ code, output = self.main(['update', '--check'])
+ self.assertEqual(code, 0)
+ self.assertIn('Stored runtime notice dismissal does not hide this availability result.', output)
+ self.assertIn('Selected version: ' + selected, output)
+ self.assertNotIn('this available release is dismissed', output)
+
+ def test_explicit_check_exposes_invalid_preference_and_never_marks_current_dismissed(self):
+ self.latest.side_effect = None
+ self.latest.return_value = ('v2.0.0', 'offline-fixture-url')
+ self.read.return_value = {'status': 'invalid', 'dismissedVersion': None,
+ 'reason': 'runtime_notice_preference_schema'}
+ code, output = self.main(['update', '--check', '--json'])
+ result = json.loads(output)
+ self.assertEqual(code, 0)
+ self.assertTrue(result['outdated'])
+ self.assertFalse(result['runtimeNoticeDismissed'])
+ self.assertEqual(result['runtimeNoticePreference']['status'], 'invalid')
+ self.read.return_value = {'status': 'set', 'dismissedVersion': '1.0.0'}
+ self.latest.return_value = ('v1.0.0', 'offline-fixture-url')
+ code, output = self.main(['update', '--check', '--json'])
+ result = json.loads(output)
+ self.assertEqual(code, 0)
+ self.assertFalse(result['outdated'])
+ self.assertFalse(result['runtimeNoticeDismissed'])
+
+ def test_explicit_preference_operation_still_runs_with_global_opt_out(self):
+ with mock.patch.dict(os.environ, {self.peer.UPDATE_NOTICE_ENV: '1'}):
+ code, output = self.main(['update', '--dismiss-runtime-notice', '2.0.0',
+ '--no-update-notice', '--json'], ambient=True)
+ self.assertEqual(code, 0)
+ self.assertEqual(json.loads(output)['dismissedVersion'], '2.0.0')
+ self.write.assert_called_once_with('2.0.0')
+ self.read.assert_not_called()
+
+ def test_refused_preference_write_preserves_normal_error_shape_and_no_update(self):
+ self.write.side_effect = self.peer.CcPeerError('Runtime notice preference refused',
+ {'reason': 'runtime_notice_preference_unsafe', 'updated': False})
+ code, output = self.main(['update', '--dismiss-runtime-notice', '2.0.0', '--json'], ambient=True)
+ result = json.loads(output)
+ self.assertEqual(code, self.peer.EXIT_ERROR)
+ self.assertFalse(result['updated'])
+ self.assertEqual(result['reason'], 'runtime_notice_preference_unsafe')
+ self.assertEqual(result['command'], 'update')
+ self.latest.assert_not_called()
+ self.distribution.assert_not_called()
+
+ def test_multi_host_reads_once_and_remote_calls_keep_notice_opt_out(self):
+ self.read.return_value = {'status': 'set', 'dismissedVersion': '2.0.0'}
+ with self.cache(), mock.patch.object(self.peer, 'tailscale_status', return_value={}), \
+ mock.patch.object(self.peer, 'resolve_ssh_destination', side_effect=lambda host, status: host), \
+ mock.patch.object(self.peer, 'tailscale_ssh_options', return_value=[]), \
+ mock.patch.object(self.peer, 'ssh_user_metadata', return_value={}), \
+ mock.patch.object(self.peer, 'run_remote', return_value={'sessions': []}) as remote, \
+ mock.patch.object(self.peer, 'remote_installed_version', return_value='1.0.0'):
+ code, output = self.main(['list', '--host', 'one', '--host', 'two', '--json'], ambient=True)
+ self.assertEqual(code, 0)
+ self.assertEqual(self.read.call_count, 1)
+ self.assertEqual(remote.call_count, 2)
+ self.assertTrue(all('--no-update-notice' in call.args[1] for call in remote.call_args_list))
+ self.assertTrue(all('clientUpdate' not in item for item in json.loads(output)))
+
+ def test_remote_check_keeps_remote_version_comparison_without_preference_read(self):
+ with mock.patch.object(self.peer, 'tailscale_status', return_value={}), \
+ mock.patch.object(self.peer, 'resolve_ssh_destination', side_effect=lambda host, status: host), \
+ mock.patch.object(self.peer, 'tailscale_ssh_options', return_value=[]), \
+ mock.patch.object(self.peer, 'ssh_user_metadata', return_value={}), \
+ mock.patch.object(self.peer, 'remote_installed_version', return_value='0.9.0'):
+ code, output = self.main(['update', '--check', '--host', 'one', '--json'])
+ result = json.loads(output)
+ self.assertEqual(code, 0)
+ self.assertTrue(result['outdated'])
+ self.assertEqual(result['remoteVersion'], '0.9.0')
+ self.assertNotIn('runtimeNoticePreference', result)
+ self.read.assert_not_called()
+
+ def test_skill_notices_and_primary_command_result_are_unchanged(self):
+ skill = {'location': 'agents', 'current': '1.0.0', 'latest': '2.0.0', 'command': 'owning-manager'}
+ self.read.side_effect = OSError('fixture preference unavailable')
+ with self.cache(), mock.patch.object(self.peer, 'prepare_skill_updates', return_value=[skill]), \
+ mock.patch.object(self.peer, 'collect_listing', return_value={'ok': True, 'sessions': []}), \
+ mock.patch.object(self.peer, 'render_listing', return_value='fixture listing'):
+ code, output = self.main(['list', '--json'], ambient=True)
+ result = json.loads(output)
+ self.assertEqual(code, 0)
+ self.assertEqual(result['sessions'], [])
+ self.assertIn('clientUpdate', result)
+ self.assertEqual(result['skillUpdates'], [skill])
+
+ def test_real_cli_remote_check_reads_only_calling_client_notice_preference(self):
+ # The client advisory is independent of the remote result: no per-host
+ # preference read or remote-version suppression is introduced.
+ self.read.return_value = {'status': 'set', 'dismissedVersion': '2.0.0'}
+ with self.cache(), mock.patch.object(self.peer, 'tailscale_status', return_value={}), \
+ mock.patch.object(self.peer, 'resolve_ssh_destination', side_effect=lambda host, status: host), \
+ mock.patch.object(self.peer, 'tailscale_ssh_options', return_value=[]), \
+ mock.patch.object(self.peer, 'ssh_user_metadata', return_value={}), \
+ mock.patch.object(self.peer, 'remote_installed_version', return_value='0.9.0') as remote:
+ code, output = self.main(['update', '--check', '--host', 'one', '--host', 'two', '--json'], ambient=True)
+ result = json.loads(output)
+ self.assertEqual(code, 0)
+ self.assertEqual(len(result), 2)
+ self.read.assert_called_once_with()
+ self.assertEqual(remote.call_count, 2)
+ for item in result:
+ self.assertTrue(item['outdated'])
+ self.assertEqual(item['remoteVersion'], '0.9.0')
+ self.assertNotIn('runtimeNoticePreference', item)
+ self.assertNotIn('clientUpdate', item)
diff --git a/tests/core/test_update_preferences.py b/tests/core/test_update_preferences.py
new file mode 100644
index 0000000..8bad4f6
--- /dev/null
+++ b/tests/core/test_update_preferences.py
@@ -0,0 +1,340 @@
+"""Isolated preference storage; metadata-controlled positives are not native QA."""
+import argparse
+import concurrent.futures
+import contextlib
+import errno
+import json
+import os
+from pathlib import Path
+import stat
+import sys
+import tempfile
+import threading
+from types import SimpleNamespace
+import unittest
+from unittest import mock
+
+import session_peer as peer
+
+
+class RuntimeNoticeVersions(unittest.TestCase):
+ def test_strict_bounded_ascii_versions(self):
+ for value, expected in (("1.2.3", "1.2.3"), ("v0.0.0", "0.0.0"), ("V12.30.4", "12.30.4")):
+ self.assertEqual(peer.runtime_notice_cli_version(value), expected)
+ self.assertEqual(peer.runtime_notice_cli_version(str(sys.maxsize)+".0.0"), str(sys.maxsize)+".0.0")
+ for value in (None, True, 123, "1.2", "01.2.3", "1.2.3-rc1", "1.2.3+meta", "１.2.3", "1.٢.3",
+ " 1.2.3", "1.2.3\n", "vV1.2.3", "9"*10000, str(sys.maxsize+1)+".0.0"):
+ with self.subTest(value_type=type(value).__name__), self.assertRaises(argparse.ArgumentTypeError):
+ peer.runtime_notice_cli_version(value)
+
+ def test_unsupported_never_touches_paths_or_posix_storage(self):
+ with mock.patch.object(peer.sys, "platform", "win32"), \
+ mock.patch.object(peer, "_runtime_notice_directory", side_effect=AssertionError("storage contacted")):
+ self.assertEqual(peer.read_runtime_notice_preference()["status"], "unsupported")
+ with self.assertRaises(peer.CcPeerError) as error:
+ peer.write_runtime_notice_preference("1.2.3")
+ self.assertEqual(error.exception.details, {"reason": "runtime_notice_preference_unsupported", "updated": False})
+
+
+@unittest.skipUnless(sys.platform in ("linux", "darwin"), "Mac/Linux preference storage")
+class RuntimeNoticePreferences(unittest.TestCase):
+ def setUp(self):
+ directory = tempfile.TemporaryDirectory(prefix="codex-update-pref-", dir=os.environ.get("SESSION_PEER_TEST_TMP"))
+ self.addCleanup(directory.cleanup)
+ # macOS temporary roots may use /var, an alias for /private/var.
+ # Positive fixtures need a canonical path; production still refuses
+ # explicitly configured paths that traverse symlink ancestors.
+ self.root = Path(directory.name).resolve()
+ self.config = self.root / "config"
+ self.leaf = self.config / "session-peer"
+ self.path = self.leaf / "update-notices.json"
+ environment = mock.patch.dict(os.environ, {"XDG_CONFIG_HOME": str(self.config)})
+ environment.start()
+ self.addCleanup(environment.stop)
+
+ @contextlib.contextmanager
+ def controlled_metadata(self):
+ # Kingston may ignore ownership/mode. Only metadata is controlled;
+ # actual directory-relative file I/O, inode binding, locks and rename run.
+ original_stat, original_fstat = os.stat, os.fstat
+ def safe(info):
+ values = {name: getattr(info, name) for name in
+ ("st_mode", "st_uid", "st_nlink", "st_dev", "st_ino", "st_size")}
+ values["st_uid"] = os.getuid()
+ values["st_mode"] = stat.S_IFMT(info.st_mode) | (0o700 if stat.S_ISDIR(info.st_mode) else 0o600)
+ return SimpleNamespace(**values)
+ with mock.patch.object(peer.os, "stat", side_effect=lambda *a, **kw: safe(original_stat(*a, **kw))), \
+ mock.patch.object(peer.os, "fstat", side_effect=lambda *a, **kw: safe(original_fstat(*a, **kw))):
+ yield
+
+ def put(self, raw):
+ self.leaf.mkdir(mode=0o700, parents=True, exist_ok=True)
+ self.path.write_bytes(raw)
+ self.path.chmod(0o600)
+
+ def test_missing_read_has_no_effect(self):
+ with self.controlled_metadata():
+ self.assertEqual(peer.read_runtime_notice_preference(), {"status": "missing", "dismissedVersion": None})
+ self.assertFalse(self.config.exists())
+
+ def test_unset_or_empty_xdg_uses_only_isolated_home_config(self):
+ for index, environment in enumerate(({}, {"XDG_CONFIG_HOME": ""})):
+ home = self.root / ("home-" + str(index))
+ with mock.patch.object(peer.os, "environ", environment), \
+ mock.patch.object(peer.Path, "home", return_value=home), self.controlled_metadata():
+ self.assertEqual(peer.read_runtime_notice_preference()["status"], "missing")
+ peer.write_runtime_notice_preference("1.2.3")
+ self.assertTrue((home / ".config/session-peer/update-notices.json").is_file())
+ self.assertFalse(self.config.exists())
+
+ def test_set_same_clear_and_stable_lock(self):
+ with self.controlled_metadata():
+ self.assertEqual(peer.write_runtime_notice_preference("v1.2.3"),
+ {"status": "set", "dismissedVersion": "1.2.3", "changed": True})
+ lock = (self.leaf / "update-notices.lock").stat().st_ino
+ before = self.path.stat().st_ino
+ self.assertFalse(peer.write_runtime_notice_preference("1.2.3")["changed"])
+ self.assertEqual(self.path.stat().st_ino, before)
+ self.assertEqual(peer.read_runtime_notice_preference(), {"status": "set", "dismissedVersion": "1.2.3"})
+ self.assertTrue(peer.write_runtime_notice_preference(None)["changed"])
+ self.assertEqual(peer.read_runtime_notice_preference(), {"status": "clear", "dismissedVersion": None})
+ self.assertFalse(peer.write_runtime_notice_preference(None)["changed"])
+ self.assertEqual((self.leaf / "update-notices.lock").stat().st_ino, lock)
+ self.assertEqual(json.loads(self.path.read_text()), {"schemaVersion": 1, "runtime": {"dismissedVersion": None}})
+
+ def test_malformed_future_or_noncanonical_state_never_suppresses_or_repairs(self):
+ invalid = (b"{", b"[]", b'{"schemaVersion":2,"runtime":{"dismissedVersion":null}}',
+ b'{"schemaVersion":true,"runtime":{"dismissedVersion":null}}',
+ b'{"schemaVersion":1,"schemaVersion":1,"runtime":{"dismissedVersion":null}}',
+ b'{"schemaVersion":1,"runtime":{"dismissedVersion":null,"extra":0}}',
+ b'{"schemaVersion":1,"runtime":{"dismissedVersion":"v1.2.3"}}',
+ b'{"schemaVersion":1,"runtime":{"dismissedVersion":NaN}}', b"x"*4097)
+ for raw in invalid:
+ self.put(raw)
+ before = self.path.stat().st_ino
+ with self.controlled_metadata():
+ self.assertEqual(peer.read_runtime_notice_preference()["status"], "invalid")
+ for version in ("2.3.4", None):
+ with self.assertRaises(peer.CcPeerError) as error:
+ peer.write_runtime_notice_preference(version)
+ self.assertEqual(error.exception.details["reason"], "runtime_notice_preference_invalid")
+ self.assertTrue(self.path.read_bytes() == raw, "invalid user state was overwritten")
+ self.assertEqual(self.path.stat().st_ino, before)
+
+ def test_unsafe_configured_paths_refuse_without_fallback(self):
+ for configured in ("relative/config", "./config", "/a/../b", "/a/./b", "/bad\x00path"):
+ with mock.patch.object(peer.os, "environ", {"XDG_CONFIG_HOME": configured}):
+ self.assertEqual(peer.read_runtime_notice_preference()["reason"], "runtime_notice_preference_unsafe")
+ with self.assertRaises(peer.CcPeerError):
+ peer.write_runtime_notice_preference("1.2.3")
+ self.assertFalse(self.config.exists())
+
+ def test_owner_modes_and_sticky_ancestor_rules(self):
+ uid = os.getuid()
+ info = lambda mode, owner=uid, links=1: SimpleNamespace(st_mode=mode, st_uid=owner, st_nlink=links)
+ peer._runtime_notice_directory_stat(info(stat.S_IFDIR | 0o755))
+ peer._runtime_notice_directory_stat(info(stat.S_IFDIR | 0o1777, 0))
+ peer._runtime_notice_directory_stat(info(stat.S_IFDIR | 0o700), private=True)
+ peer._runtime_notice_directory_stat(info(stat.S_IFDIR | 0o2700), private=True)
+ peer._runtime_notice_file_stat(info(stat.S_IFREG | 0o600))
+ for value, private in ((info(stat.S_IFDIR | 0o777), False),
+ (info(stat.S_IFDIR | 0o700, uid+1000), False),
+ (info(stat.S_IFDIR | 0o755), True), (info(stat.S_IFDIR | 0o500), True)):
+ with self.assertRaises(peer.RuntimeNoticePreferenceFailure):
+ peer._runtime_notice_directory_stat(value, private=private)
+ for value in (info(stat.S_IFREG | 0o644), info(stat.S_IFREG | 0o600, uid+1000),
+ info(stat.S_IFREG | 0o600, links=2), info(stat.S_IFIFO | 0o600)):
+ with self.assertRaises(peer.RuntimeNoticePreferenceFailure):
+ peer._runtime_notice_file_stat(value)
+
+ def test_symlink_and_fifo_targets_refuse_without_following_or_blocking(self):
+ target = self.root / "other"
+ target.write_text("untouched")
+ self.leaf.mkdir(parents=True, mode=0o700)
+ self.path.symlink_to(target)
+ with self.controlled_metadata():
+ self.assertEqual(peer.read_runtime_notice_preference()["reason"], "runtime_notice_preference_unsafe")
+ with self.assertRaises(peer.CcPeerError):
+ peer.write_runtime_notice_preference("1.2.3")
+ self.assertEqual(target.read_text(), "untouched")
+ self.path.unlink()
+ try:
+ os.mkfifo(self.path, mode=0o600)
+ except OSError as exc:
+ if exc.errno in (errno.ENOTSUP, errno.EOPNOTSUPP):
+ self.skipTest("fixture filesystem has no FIFO support")
+ raise
+ with self.controlled_metadata():
+ self.assertEqual(peer.read_runtime_notice_preference()["reason"], "runtime_notice_preference_unsafe")
+
+ def test_symlink_ancestor_and_stable_lock_refuse(self):
+ alternate = self.root / "alternate"
+ alternate.mkdir(mode=0o700)
+ self.config.symlink_to(alternate, target_is_directory=True)
+ with self.controlled_metadata():
+ self.assertEqual(peer.read_runtime_notice_preference()["reason"], "runtime_notice_preference_unsafe")
+ with self.assertRaises(peer.CcPeerError):
+ peer.write_runtime_notice_preference("1.2.3")
+ self.assertEqual(list(alternate.iterdir()), [])
+ self.config.unlink()
+ self.leaf.mkdir(mode=0o700, parents=True)
+ target = self.root / "untouched-lock"
+ target.write_text("untouched")
+ (self.leaf / "update-notices.lock").symlink_to(target)
+ with self.controlled_metadata():
+ with self.assertRaises(peer.CcPeerError) as error:
+ peer.write_runtime_notice_preference("1.2.3")
+ self.assertEqual(error.exception.details["reason"], "runtime_notice_preference_unsafe")
+ self.assertEqual(target.read_text(), "untouched")
+ self.assertFalse(self.path.exists())
+
+ def test_temporary_root_alias_refuses_while_resolved_fixture_succeeds(self):
+ actual = self.root / "actual-root"
+ actual.mkdir(mode=0o700)
+ alias = self.root / "root-alias"
+ alias.symlink_to(actual, target_is_directory=True)
+ configured = alias / "config"
+ with self.controlled_metadata():
+ with mock.patch.dict(os.environ, {"XDG_CONFIG_HOME": str(configured)}):
+ self.assertEqual(peer.read_runtime_notice_preference()["reason"],
+ "runtime_notice_preference_unsafe")
+ with self.assertRaises(peer.CcPeerError) as error:
+ peer.write_runtime_notice_preference("1.2.3")
+ self.assertEqual(error.exception.details["reason"], "runtime_notice_preference_unsafe")
+ self.assertEqual(list(actual.iterdir()), [])
+ with mock.patch.dict(os.environ, {"XDG_CONFIG_HOME": str(configured.resolve())}):
+ self.assertEqual(peer.read_runtime_notice_preference()["status"], "missing")
+ self.assertTrue(peer.write_runtime_notice_preference("1.2.3")["changed"])
+ self.assertEqual(peer.read_runtime_notice_preference()["dismissedVersion"], "1.2.3")
+
+ def test_inode_substitution_is_refused(self):
+ self.put(b'{"schemaVersion":1,"runtime":{"dismissedVersion":"1.2.3"}}')
+ with self.controlled_metadata():
+ original = peer.os.fstat
+ def substituted(fd):
+ info = original(fd)
+ if stat.S_ISREG(info.st_mode):
+ fields = dict(vars(info))
+ fields["st_ino"] += 1
+ return SimpleNamespace(**fields)
+ return info
+ with mock.patch.object(peer.os, "fstat", side_effect=substituted):
+ self.assertEqual(peer.read_runtime_notice_preference()["reason"], "runtime_notice_preference_unsafe")
+
+ def test_short_regular_reads_are_completed_within_byte_cap(self):
+ self.put(b'{"schemaVersion":1,"runtime":{"dismissedVersion":"1.2.3"}}')
+ original = os.read
+ with self.controlled_metadata(), mock.patch.object(peer.os, "read", side_effect=lambda fd, size: original(fd, min(size, 3))):
+ self.assertEqual(peer.read_runtime_notice_preference()["dismissedVersion"], "1.2.3")
+
+ def test_pre_replace_failure_preserves_previous_complete_preference_and_cleans_temp(self):
+ with self.controlled_metadata():
+ peer.write_runtime_notice_preference("1.2.3")
+ before = self.path.read_bytes()
+ with mock.patch.object(peer.os, "replace", side_effect=OSError("private diagnostic")):
+ with self.assertRaises(peer.CcPeerError) as error:
+ peer.write_runtime_notice_preference("2.3.4")
+ self.assertEqual(str(error.exception), "Runtime notice preference refused")
+ self.assertEqual(error.exception.details["reason"], "runtime_notice_preference_write_failed")
+ self.assertNotIn("preferenceMayHaveChanged", error.exception.details)
+ self.assertTrue(self.path.read_bytes() == before, "failed atomic replacement changed preference")
+ self.assertEqual(sorted(path.name for path in self.leaf.iterdir()), ["update-notices.json", "update-notices.lock"])
+
+ def test_post_replace_durability_failure_does_not_claim_no_effect(self):
+ original = os.fsync
+ def sync(fd):
+ if stat.S_ISDIR(os.fstat(fd).st_mode):
+ raise OSError("private diagnostic")
+ return original(fd)
+ with self.controlled_metadata():
+ peer.write_runtime_notice_preference("1.2.3")
+ with mock.patch.object(peer.os, "fsync", side_effect=sync):
+ with self.assertRaises(peer.CcPeerError) as error:
+ peer.write_runtime_notice_preference("2.3.4")
+ self.assertTrue(error.exception.details["preferenceMayHaveChanged"])
+ self.assertFalse(error.exception.details["updated"])
+ self.assertEqual(json.loads(self.path.read_text())["runtime"]["dismissedVersion"], "2.3.4")
+
+ def test_owned_temporary_validation_failure_cleans_only_bound_inode(self):
+ with self.controlled_metadata():
+ peer.write_runtime_notice_preference("1.2.3")
+ before = self.path.read_bytes()
+ original_open, original_validate = peer.os.open, peer._runtime_notice_file_stat
+ temporary_inodes = set()
+ def opening(name, *args, **kwargs):
+ fd = original_open(name, *args, **kwargs)
+ if isinstance(name, str) and name.startswith(".update-notices-"):
+ temporary_inodes.add(peer.os.fstat(fd).st_ino)
+ return fd
+ def validating(info):
+ if info.st_ino in temporary_inodes:
+ peer._runtime_notice_fail("unsafe")
+ return original_validate(info)
+ with mock.patch.object(peer.os, "open", side_effect=opening), \
+ mock.patch.object(peer, "_runtime_notice_file_stat", side_effect=validating):
+ with self.assertRaises(peer.CcPeerError) as error:
+ peer.write_runtime_notice_preference("2.3.4")
+ self.assertEqual(error.exception.details["reason"], "runtime_notice_preference_unsafe")
+ self.assertEqual(self.path.read_bytes(), before)
+ self.assertEqual(sorted(path.name for path in self.leaf.iterdir()), ["update-notices.json", "update-notices.lock"])
+
+ def test_lock_timeout_uses_stable_file_and_original_budget(self):
+ clock = [0.0]
+ def sleep(seconds):
+ clock[0] += seconds
+ with self.controlled_metadata(), \
+ mock.patch.object(peer.time, "monotonic", side_effect=lambda: clock[0]), \
+ mock.patch.object(peer.time, "sleep", side_effect=sleep), \
+ mock.patch.object(peer.fcntl, "flock", side_effect=BlockingIOError()):
+ with self.assertRaises(peer.CcPeerError) as error:
+ peer.write_runtime_notice_preference("1.2.3")
+ self.assertEqual(error.exception.details["reason"], "runtime_notice_preference_busy")
+ self.assertLessEqual(clock[0], 2.0)
+ self.assertTrue((self.leaf / "update-notices.lock").exists())
+ self.assertFalse(self.path.exists())
+
+ def test_real_flock_concurrent_writers_leave_one_complete_preference(self):
+ barrier = threading.Barrier(2)
+ def write(version):
+ barrier.wait(timeout=2)
+ return peer.write_runtime_notice_preference(version)
+ with self.controlled_metadata():
+ # Create and pin the dedicated directory/lock before concurrency.
+ peer.write_runtime_notice_preference(None)
+ with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor:
+ results = list(executor.map(write, ("1.2.3", "2.3.4")))
+ current = peer.read_runtime_notice_preference()
+ self.assertTrue(all(result["changed"] for result in results))
+ self.assertIn(current["dismissedVersion"], ("1.2.3", "2.3.4"))
+
+ def test_concurrent_initial_writers_share_one_new_stable_lock(self):
+ barrier = threading.Barrier(2)
+ def write(version):
+ barrier.wait(timeout=2)
+ return peer.write_runtime_notice_preference(version)
+ with self.controlled_metadata():
+ with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor:
+ results = list(executor.map(write, ("1.2.3", "2.3.4")))
+ current = peer.read_runtime_notice_preference()
+ self.assertEqual(len(results), 2)
+ self.assertIn(current["dismissedVersion"], ("1.2.3", "2.3.4"))
+ self.assertEqual(sorted(path.name for path in self.leaf.iterdir()), ["update-notices.json", "update-notices.lock"])
+
+ def test_ambient_unexpected_exception_is_fixed_invalid_without_repairs(self):
+ with mock.patch.object(peer, "_runtime_notice_directory", side_effect=RuntimeError("private diagnostic")):
+ self.assertEqual(peer.read_runtime_notice_preference(),
+ {"status": "invalid", "dismissedVersion": None, "reason": "runtime_notice_preference_unavailable"})
+ self.assertFalse(self.config.exists())
+
+ def test_native_private_path_when_filesystem_enforces_metadata(self):
+ state = peer.read_runtime_notice_preference()
+ if state["status"] == "invalid" and state.get("reason") == "runtime_notice_preference_unsafe":
+ self.skipTest("native task volume ancestors fail ownership/mode policy; controlled positives are not native certification")
+ self.assertEqual(state["status"], "missing")
+ self.assertTrue(peer.write_runtime_notice_preference("1.2.3")["changed"])
+ self.assertEqual(stat.S_IMODE(self.leaf.stat().st_mode), 0o700)
+ self.assertEqual(stat.S_IMODE(self.path.stat().st_mode), 0o600)
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/tests/core/test_updates.py b/tests/core/test_updates.py
index 4b96daf..8f9bddf 100644
--- a/tests/core/test_updates.py
+++ b/tests/core/test_updates.py
@@ -36,6 +36,10 @@ def setUp(self):
skill_patch = mock.patch.object(session_peer, "installed_skills", return_value=[])
skill_patch.start()
self.addCleanup(skill_patch.stop)
+ preference_patch = mock.patch.object(session_peer, "read_runtime_notice_preference",
+ return_value={"status": "missing", "dismissedVersion": None})
+ preference_patch.start()
+ self.addCleanup(preference_patch.stop)
@staticmethod
def args(**overrides):
@@ -357,6 +361,12 @@ def test_background_refresh_populates_both_caches(self):
class PackageManagement(unittest.TestCase):
+ def setUp(self):
+ preference_patch = mock.patch.object(session_peer, "read_runtime_notice_preference",
+ return_value={"status": "missing", "dismissedVersion": None})
+ preference_patch.start()
+ self.addCleanup(preference_patch.stop)
+
def test_package_update_does_not_download(self):
import contextlib
import io
diff --git a/tools/generate_session_peer.py b/tools/generate_session_peer.py
index 4817035..6deaa08 100644
--- a/tools/generate_session_peer.py
+++ b/tools/generate_session_peer.py
@@ -32,6 +32,7 @@
"messages.py",
"release_verification.py",
"updates.py",
+ "update_preferences.py",
"handoff_stdin.py",
"handoff_process.py",
"handoff.py",