diff --git a/frontend-web/src/components/workspace/settings/skill-settings-page.tsx b/frontend-web/src/components/workspace/settings/skill-settings-page.tsx
index 2a011b25..67db57a4 100644
--- a/frontend-web/src/components/workspace/settings/skill-settings-page.tsx
+++ b/frontend-web/src/components/workspace/settings/skill-settings-page.tsx
@@ -2,6 +2,7 @@
 
 import {
   AlertCircle,
+  Anchor,
   ArchiveIcon,
   ArrowDownUpIcon,
   BookmarkIcon,
@@ -77,6 +78,7 @@ import {
   useGenerateSkillNameZh,
   useInstallSkillUpload,
   useReconcileCustomSkills,
+  useSetSkillAlwaysOn,
   useSetSkillFeatured,
   useSetSkillNameZh,
   useSetSkillPinned,
@@ -84,7 +86,11 @@ import {
   useSkills,
   useValidateSkillUpload,
 } from "@/core/skills/hooks";
-import { usePublishSquares, useViewMode } from "@/core/system-settings/hooks";
+import {
+  usePublishSquares,
+  useSkillCompressionSettings,
+  useViewMode,
+} from "@/core/system-settings/hooks";
 import type { Skill } from "@/core/skills/type";
 import { formatDateTime } from "@/core/utils/datetime";
 import { env } from "@/env";
@@ -181,8 +187,10 @@ function SkillSettingsList({ onClose }: { onClose?: () => void }) {
   const isAdmin = storedAuth?.system_role === "admin";
 
   // Name + tag filtering is resolved server-side via the /api/skills query.
-  // Non-admins get their 岗位 (position) visibility applied (scope=mine): a
-  // position that binds skill tags restricts which skills they can see.
+  // A 岗位 (position) no longer *restricts* the list — it grants skills into the
+  // member's 我的技能 (each such skill comes back with ``favorited: true``), so
+  // the 广场 always shows every visible skill and the 我的 tab includes the
+  // position-granted ones. ``scope`` is kept for API compatibility only.
   const { skills, isLoading, error } = useSkills({
     search,
     tagIds,
@@ -217,7 +225,9 @@ function SkillSettingsList({ onClose }: { onClose?: () => void }) {
       if (isBuiltin || !!skill.published) {
         all.push(skill);
       }
-      if (isOwnedByMe) {
+      // "我的" = skills I own OR skills granted into my collection by my 岗位
+      // (position) — mirroring how favorited agents land in 我的智能体.
+      if (isOwnedByMe || !!skill.favorited) {
         mine.push(skill);
       }
     }
@@ -601,6 +611,8 @@ export function SkillRow({
   const { data: cardPublishSquares } = usePublishSquares();
   const setPinned = useSetSkillPinned();
   const setFeatured = useSetSkillFeatured();
+  const setAlwaysOn = useSetSkillAlwaysOn();
+  const { data: compression } = useSkillCompressionSettings();
   const deleteSkill = useDeleteCustomSkill();
   const duplicateSkillMutation = useDuplicateSkill();
   const [confirmDeleteOpen, setConfirmDeleteOpen] = useState(false);
@@ -620,6 +632,17 @@ export function SkillRow({
   // Admin can edit/delete/unpublish anyone's skill; non-admin only their own.
   const canManage = isCustom && (isOwned || isAdmin);
   const canTogglePublish = canManage && !isBuiltin;
+  // 常驻(免压缩):内置技能仅管理员可设;自定义技能其所有者或管理员可设(与后端一致)。
+  const canSetAlwaysOn = isAdmin || isOwned;
+
+  async function handleToggleAlwaysOn(next: boolean) {
+    try {
+      await setAlwaysOn.mutateAsync({ skillName: skill.name, alwaysOn: next });
+      toast.success(next ? "已设为常驻(免压缩)" : "已取消常驻");
+    } catch (err) {
+      toast.error(err instanceof Error ? err.message : String(err));
+    }
+  }
 
   async function handleTogglePublished(next: boolean) {
     // Publishing: prompt for which square to publish into (when configured).
@@ -932,6 +955,27 @@ export function SkillRow({
           />
         </Button>
       )}
+      {compression?.enabled && canSetAlwaysOn && (
+        <Button
+          size="icon"
+          variant="ghost"
+          className={cn(
+            "h-8 w-8 shrink-0",
+            skill.always_on && "text-amber-600 hover:text-amber-700",
+          )}
+          disabled={setAlwaysOn.isPending}
+          onClick={() => void handleToggleAlwaysOn(!skill.always_on)}
+          title={
+            skill.always_on
+              ? "取消常驻(将被压缩,改由 search_skills 检索)"
+              : "设为常驻(免压缩,始终全文注入提示词)"
+          }
+        >
+          <Anchor
+            className={cn("size-3.5", skill.always_on && "fill-current")}
+          />
+        </Button>
+      )}
       {canTogglePublish && (
         <Button
           size="icon"
diff --git a/frontend-web/src/core/positions/types.ts b/frontend-web/src/core/positions/types.ts
index 07b2a4ee..3e3ff5f1 100644
--- a/frontend-web/src/core/positions/types.ts
+++ b/frontend-web/src/core/positions/types.ts
@@ -24,7 +24,7 @@ export const POSITION_RESOURCE_LABELS: Record<PositionResourceType, string> = {
 
 export const POSITION_RESOURCE_HINT: Record<PositionResourceType, string> = {
   agent: "不配置标签 = 默认加载全部内置智能体；配置后只加载所选标签下的智能体。",
-  skill: "不配置标签 = 可见全部技能；配置后只可见所选标签下的技能。",
+  skill: "不配置标签 = 不向「我的技能」添加任何技能；配置后将所选标签下的技能加入成员的「我的技能」（技能广场仍展示全部技能）。",
   scheduled_task: "不配置标签 = 不订阅任何定时任务；配置后订阅所选标签下的定时任务。",
   recommended_question: "不配置标签 = 展示全部推荐问题；配置后只展示所选标签下的推荐问题（聊天框上方的推荐问题）。",
 };
diff --git a/frontend-web/src/core/skills/type.ts b/frontend-web/src/core/skills/type.ts
index 43becf88..dc0ba9da 100644
--- a/frontend-web/src/core/skills/type.ts
+++ b/frontend-web/src/core/skills/type.ts
@@ -61,6 +61,9 @@ export interface Skill {
   /** Publish square (发布广场) the skill is filed under once published. "" = default square. */
   square_id?: string;
   builtin?: boolean;
+  // True when this skill is in the caller's 我的技能 collection — e.g. granted by
+  // their 岗位 (position). Owned skills are detected separately via owner_user_id.
+  favorited?: boolean;
   // Long-form detail/introduction authored by the skill owner or an admin.
   // Distinct from ``description`` (parsed from SKILL.md); supports Markdown.
   detail?: string | null;
diff --git a/offline-backend-20260512/backend/CLAUDE.md b/offline-backend-20260512/backend/CLAUDE.md
index 81ced960..fe2fe38d 100644
--- a/offline-backend-20260512/backend/CLAUDE.md
+++ b/offline-backend-20260512/backend/CLAUDE.md
@@ -360,16 +360,24 @@ shown to users ARE the **recommended questions** (`recommended_questions` table)
   extended with `recommended_question`).
 
 **Sync engine** (`deerflow/services/position_sync.py`, wired as
-`app.state.position_sync`). `PositionSyncService.resolve_visible(user, type)`
-returns `set | None` (None = unrestricted) for the **query-time** visibility
-types (`skill`, `recommended_question`). `sync_user(user)` **materializes** the
-other two into the existing "我的" tables, idempotently:
+`app.state.position_sync`). `sync_user(user)` **materializes** three resource
+types into the existing "我的" tables, idempotently:
 - agents → `agent_favorites` rows with `origin='position'`
+- **skills → `skill_favorites` rows with `origin='position'`** (table
+  `SkillFavoriteRow`, migration `20260616_02`, auto-created by `create_all`)
 - scheduled tasks → `scheduled_task_subscriptions` rows with `origin='position'`
 
+Skills behave **exactly like agents**: a position only *adds* its tagged skills
+to the member's 我的技能 — the browse-all 广场 is never restricted. (This replaced
+the earlier query-time visibility filter, which wrongly shrank the square to the
+bound skill.) `PositionSyncService.resolve_visible(user, type)` returns
+`set | None` (None = unrestricted) and is now used **only** for the one remaining
+query-time type, `recommended_question`.
+
 Sync only ever adds/removes `origin='position'` rows, so a user's own
-favorites/subscriptions are never clobbered. Default sets: agents = built-ins
-(`user_id IS NULL`); scheduled tasks = empty; skills/questions = unrestricted.
+favorites/subscriptions are never clobbered. Default sets (no binding for a
+type): agents = built-ins (`user_id IS NULL`); skills = empty (square stays
+full, nothing added to 我的); scheduled tasks = empty; questions = unrestricted.
 `origin`/`position_id` columns were added to `agent_favorites` and
 `scheduled_task_subscriptions` — see migration `20260616_01` +
 `scripts/sql/column_additions_mysql.sql`. The running app also auto-adds these
@@ -390,8 +398,11 @@ unassigned users inherit, else `null`); powers the top-bar 岗位 label
 (`strategy-components/Header.tsx` user cluster, via `useMyPosition`). Related
 changes: `GET /api/recommended-questions` is now
 filtered by the caller's position (`?all=1` bypasses for admin management) and
-admin can tag questions via `/api/tags/assignments/recommended_question/{id}`;
-`GET /api/skills?scope=mine` applies skill visibility. Frontend: `core/positions`
+admin can tag questions via `/api/tags/assignments/recommended_question/{id}`.
+`GET /api/skills` no longer restricts by position; each skill carries a
+`favorited` flag (true = in the caller's 我的技能, e.g. position-granted) so the
+skills page's 我的 tab includes owned ∪ position-granted skills while the 广场
+stays complete. Frontend: `core/positions`
 + admin `PositionManagementPage` (侧边栏「综合管理 → 岗位管理」 + 右下角设置菜单).
 Recommended questions stay rendered **above** the chat input (their original
 position), just filtered by the caller's position. Tests:
@@ -499,11 +510,18 @@ prompt body and surfaces the rest on demand via keyword retrieval.
   default 5, `keep_index_in_prompt` default true). Read/written by
   `GET`/`PUT /api/system-settings/skill-compression` (GET open to any user — the
   skill detail page reads it; PUT admin-only, refreshes the prompt cache).
-- **Per-skill 常驻 flag** — `extensions_config.json → skills.{name}.always_on`
-  (`SkillStateConfig.always_on`, alongside `enabled`). Surfaced on
-  `Skill.always_on` (populated in `load_skills`) and `SkillResponse.always_on`;
-  toggled via `PUT /api/skills/{name}/always-on`. `update_skill` /
-  `update_skill_display` preserve it when rewriting the config entry.
+- **Per-skill 常驻 flag** — stored in the **DB**: `SkillRow.always_on`
+  (`Boolean`, migration `20260616_03`, idempotent SQL in
+  `column_additions_mysql.sql`), **not** in `extensions_config.json` (the
+  source-tree config file is read-only / unsafe to rewrite in some deployments,
+  and built-in skills carry no file entry). Threaded through `SkillStore`
+  create/ensure_legacy/update/update_admin + `list_always_on_names()` exactly
+  like `name_zh`/`detail`; surfaced on `SkillResponse.always_on` from the
+  ownership record in `_skill_to_response`. Toggled via
+  `PUT /api/skills/{name}/always-on` (custom skill → owner or admin; built-in /
+  public → admin only, materializes a row on demand via `ensure_legacy`). The
+  prompt builder stamps `Skill.always_on` from `list_always_on_names()` via a
+  sync DB bridge (`_apply_always_on` in `lead_agent/prompt.py`).
 - **Prompt split** — `get_skills_prompt_section` (`lead_agent/prompt.py`) reads
   the compression settings (skipped for restricted custom agents — they already
   have a curated allowlist). When **off**, behaviour is unchanged (all enabled
@@ -688,7 +706,7 @@ Both endpoints are whitelisted in `auth_middleware._PUBLIC_EXACT_PATHS`. The sha
 
 **`extensions_config.json`**:
 - `mcpServers` - Map of server name ? config (enabled, type, command, args, env, url, headers, oauth, description)
-- `skills` - Map of skill name ? state (enabled, always_on)
+- `skills` - Map of skill name ? state (enabled). 常驻(免压缩)/always_on lives in the `skills` DB table, not here.
 
 Both can be modified at runtime via Gateway API endpoints or `DeerFlowClient` methods.
 
diff --git a/offline-backend-20260512/backend/app/gateway/routers/skills.py b/offline-backend-20260512/backend/app/gateway/routers/skills.py
index 61c0507f..dad9758f 100644
--- a/offline-backend-20260512/backend/app/gateway/routers/skills.py
+++ b/offline-backend-20260512/backend/app/gateway/routers/skills.py
@@ -169,6 +169,7 @@ class SkillResponse(BaseModel):
     published: bool = Field(default=False, description="Whether the owner has made this skill visible to other users")
     square_id: str = Field(default="", description="Publish square (发布广场) the skill is filed under once published. Empty string = the default square.")
     builtin: bool = Field(default=False, description="True when owner_user_id is null (no human owner)")
+    favorited: bool = Field(default=False, description="True when this skill is in the caller's 我的技能 collection (e.g. granted by their 岗位/position). Owned skills are detected separately via owner_user_id.")
     detail: str | None = Field(default=None, description="Human-authored long-form detail text (Markdown), editable by owner/admin")
     display: dict[str, Any] | None = Field(default=None, description="Optional frontend display adapter")
     display_sample: dict[str, Any] | None = Field(default=None, alias="displaySample", description="Persisted sample output used by the display adapter editor")
@@ -280,6 +281,7 @@ def _skill_to_response(
     *,
     display_override: dict[str, Any] | None | object = _UNSET,
     display_sample: dict[str, Any] | None = None,
+    favorited: bool = False,
 ) -> SkillResponse:
     """Convert a Skill object to a SkillResponse.
 
@@ -317,6 +319,9 @@ def _skill_to_response(
     detail: str | None = None
     name_zh: str | None = None
     square_id = ""
+    # 常驻(免压缩) is stored in the DB (SkillRow.always_on), surfaced via the
+    # ownership record — never read from the on-disk extensions_config.json.
+    always_on = False
     if ownership is not None:
         owner_user_id = ownership.get("owner_user_id")
         published = bool(ownership.get("published", False))
@@ -324,6 +329,7 @@ def _skill_to_response(
         detail = ownership.get("detail")
         name_zh = ownership.get("name_zh")
         square_id = str(ownership.get("square_id") or "")
+        always_on = bool(ownership.get("always_on", False))
 
     display = None if display_override is _UNSET else display_override
     if display_override is _UNSET:
@@ -341,7 +347,7 @@ def _skill_to_response(
         license=skill.license,
         category=skill.category,
         enabled=skill.enabled,
-        always_on=skill.always_on,
+        always_on=always_on,
         use_count=entry.get("use_count", 0),
         view_count=entry.get("view_count", 0),
         patch_count=entry.get("patch_count", 0),
@@ -355,6 +361,7 @@ def _skill_to_response(
         published=published,
         square_id=square_id,
         builtin=builtin,
+        favorited=favorited,
         detail=detail,
         display=display,
         displaySample=display_sample,
@@ -810,6 +817,15 @@ async def list_skills(
             ownership = {record["name"]: record for record in ownership_records}
         else:
             ownership = await _build_ownership_index(skill_store, user_id)
+        # Skills the caller's 岗位 (position) granted into their "我的技能"
+        # collection. Admins browse everything, so the badge is moot for them.
+        favorite_names: set[str] = set()
+        if not admin:
+            try:
+                favorite_names = await skill_store.list_favorite_skill_names(user_id)
+            except Exception:
+                logger.debug("skill favorites lookup failed", exc_info=True)
+                favorite_names = set()
         display_adapters = await _list_skill_display_adapters([skill.name for skill in all_skills])
         responses: list[SkillResponse] = []
         for skill in all_skills:
@@ -825,22 +841,30 @@ async def list_skills(
                         ownership.get(skill.name),
                         display_override=_display_override_from_adapter(adapter),
                         display_sample=adapter.get("sample") if adapter else None,
+                        favorited=skill.name in favorite_names,
                     )
                 )
                 continue
             record = ownership.get(skill.name)
             if record is None:
-                if not admin:
+                if admin:
+                    # Admin sees every custom skill on disk even without an
+                    # ownership row (legacy / orphaned), so they can clean it up.
+                    record = None
+                elif skill.name in favorite_names:
+                    # A position granted this skill even though it isn't
+                    # otherwise visible to the user (e.g. an unpublished custom
+                    # skill). Surface it so it lands in their 我的技能.
+                    record = await skill_store.get_any(skill.name)
+                else:
                     continue
-                # Admin sees every custom skill on disk even without an
-                # ownership row (legacy / orphaned), so they can clean it up.
-                record = None
             responses.append(
                 _skill_to_response(
                     skill,
                     record,
                     display_override=_display_override_from_adapter(adapter),
                     display_sample=adapter.get("sample") if adapter else None,
+                    favorited=skill.name in favorite_names,
                 )
             )
         responses = await _attach_and_filter_tags(tag_store, responses, search, tag_ids)
@@ -855,19 +879,12 @@ async def list_skills(
                 for r in responses
                 if (r.square_id or "") == square_id or (square_is_default and not (r.square_id or ""))
             ]
-        # 岗位 (position) visibility: when the caller asks for "我的技能",
-        # restrict to the skills their position grants. ``None`` from the sync
-        # service = unrestricted (no skill tag binding → all skills visible).
-        if scope == "mine" and not admin:
-            sync = getattr(request.app.state, "position_sync", None)
-            if sync is not None:
-                try:
-                    visible = await sync.resolve_visible(user_id, "skill")
-                except Exception:
-                    logger.debug("skill visibility resolve failed", exc_info=True)
-                    visible = None
-                if visible is not None:
-                    responses = [r for r in responses if r.name in visible]
+        # 岗位 (position) no longer *restricts* the skill list: a position grants
+        # skills by materializing them into the member's 我的技能 (the ``favorited``
+        # flag above + the ``skill_favorites`` table, synced by PositionSyncService),
+        # mirroring how agents are added to 我的智能体. The browse-all 广场 therefore
+        # always shows every visible skill — the ``scope`` param is kept for API
+        # compatibility but intentionally does not filter here.
         responses = _sort_skill_responses(responses, sort)
         return SkillsListResponse(skills=responses)
     except HTTPException:
@@ -2194,7 +2211,6 @@ async def update_skill(skill_name: str, request: SkillUpdateRequest, config: App
         current_skill_config = extensions_config.skills.get(skill_name)
         extensions_config.skills[skill_name] = SkillStateConfig(
             enabled=request.enabled,
-            always_on=current_skill_config.always_on if current_skill_config else skill.always_on,
             display=current_skill_config.display if current_skill_config else None,
         )
 
@@ -2229,41 +2245,50 @@ async def update_skill(skill_name: str, request: SkillUpdateRequest, config: App
     summary="Update Skill 常驻(免压缩) Flag",
     description="Mark a skill as 常驻(免压缩) so it is injected full-text into the system prompt even when skill compression is enabled.",
 )
-async def update_skill_always_on(skill_name: str, request: SkillAlwaysOnUpdateRequest, config: AppConfig = Depends(get_config)) -> SkillResponse:
+async def update_skill_always_on(
+    skill_name: str,
+    request: SkillAlwaysOnUpdateRequest,
+    http_request: Request,
+    config: AppConfig = Depends(get_config),
+    skill_store: SkillStore = Depends(get_skill_store),
+) -> SkillResponse:
+    """Persist the 常驻(免压缩) flag in the **DB** (skills table), not in
+    extensions_config.json — the source-tree config file is read-only / unsafe to
+    rewrite in some deployments, and built-in skills carry no file entry anyway.
+    Mirrors the detail/name_zh endpoints: custom skill → owner or admin; built-in
+    / public skill → admin only (materializes a row on demand)."""
     try:
         skill_name = skill_name.replace("\r\n", "").replace("\n", "")
+        user_id = _current_user_id(http_request)
+        is_admin = _is_admin_user(http_request)
+
         skills = get_or_new_skill_storage(app_config=config).load_skills(enabled_only=False)
         skill = next((s for s in skills if s.name == skill_name), None)
-
         if skill is None:
             raise HTTPException(status_code=404, detail=f"Skill '{skill_name}' not found")
 
-        config_path = ExtensionsConfig.resolve_config_path()
-        if config_path is None:
-            config_path = Path.cwd().parent / "extensions_config.json"
-            logger.info(f"No existing extensions config found. Creating new config at: {config_path}")
+        if skill.category == SkillCategory.CUSTOM:
+            _record, mode = await _get_editable_skill_or_raise(skill_store, skill_name, user_id, is_admin=is_admin)
+            if mode == "admin":
+                updated = await skill_store.update_admin(skill_name, {"always_on": request.always_on})
+            else:
+                updated = await skill_store.update(skill_name, user_id, {"always_on": request.always_on})
+        else:
+            # Built-in / public skill — admin only; no DB row by default.
+            if not is_admin:
+                raise HTTPException(status_code=403, detail="Only admin users can change a built-in skill's 常驻 flag")
+            if await skill_store.get_any(skill_name) is None:
+                await skill_store.ensure_legacy({"name": skill_name, "owner_user_id": None, "published": True})
+            updated = await skill_store.update_admin(skill_name, {"always_on": request.always_on})
 
-        extensions_config = get_extensions_config()
-        current_skill_config = extensions_config.skills.get(skill_name)
-        extensions_config.skills[skill_name] = SkillStateConfig(
-            enabled=current_skill_config.enabled if current_skill_config else skill.enabled,
-            always_on=request.always_on,
-            display=current_skill_config.display if current_skill_config else None,
-        )
+        if updated is None:
+            raise HTTPException(status_code=500, detail="Failed to persist skill 常驻 flag")
 
-        with open(config_path, "w", encoding="utf-8") as f:
-            json.dump(_dump_extensions_config(extensions_config), f, indent=2)
-
-        reload_extensions_config()
+        # always_on lives in the system prompt's skill injection — refresh the cache.
         await refresh_skills_system_prompt_cache_async()
 
-        skills = get_or_new_skill_storage(app_config=config).load_skills(enabled_only=False)
-        updated_skill = next((s for s in skills if s.name == skill_name), None)
-        if updated_skill is None:
-            raise HTTPException(status_code=500, detail=f"Failed to reload skill '{skill_name}' after update")
-
-        logger.info(f"Skill '{skill_name}' always_on updated to {request.always_on}")
-        return _skill_to_response(updated_skill)
+        logger.info("Skill '%s' always_on updated to %s", skill_name, request.always_on)
+        return _skill_to_response(skill, updated)
 
     except HTTPException:
         raise
@@ -2300,7 +2325,6 @@ async def update_skill_display(
         current_skill_config = extensions_config.skills.get(skill_name)
         extensions_config.skills[skill_name] = SkillStateConfig(
             enabled=current_skill_config.enabled if current_skill_config else skill.enabled,
-            always_on=current_skill_config.always_on if current_skill_config else skill.always_on,
             display=request.display,
         )
 
diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/agents/lead_agent/prompt.py b/offline-backend-20260512/backend/packages/harness/deerflow/agents/lead_agent/prompt.py
index 9c489d98..839a83e0 100644
--- a/offline-backend-20260512/backend/packages/harness/deerflow/agents/lead_agent/prompt.py
+++ b/offline-backend-20260512/backend/packages/harness/deerflow/agents/lead_agent/prompt.py
@@ -86,8 +86,58 @@ def _drop_archived_skills(skills: list[Skill]) -> list[Skill]:
         return skills
 
 
+def _run_coro_blocking(coro):
+    """Run an async coroutine to completion from sync code.
+
+    Mirrors ``skill_tools._run_coro_blocking``: when already inside a running
+    event loop (e.g. prompt built during an async agent run), hand the coroutine
+    to a fresh loop in a worker thread; otherwise run it directly.
+    """
+    try:
+        asyncio.get_running_loop()
+        in_loop = True
+    except RuntimeError:
+        in_loop = False
+
+    if in_loop:
+        import concurrent.futures
+
+        with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool:
+            return pool.submit(asyncio.run, coro).result()
+    return asyncio.run(coro)
+
+
+def _load_always_on_names() -> set[str]:
+    """Names marked 常驻(免压缩), read from the **DB** (skills table).
+
+    Returns an empty set on any failure / no-DB deployment, so compression then
+    simply treats every skill as compressible (safe default).
+    """
+    try:
+        from deerflow.persistence.engine import get_session_factory
+        from deerflow.persistence.skills import make_skill_store
+
+        store = make_skill_store(get_session_factory())
+        return set(_run_coro_blocking(store.list_always_on_names()))
+    except Exception:
+        logger.debug("Failed to load always_on skill names from DB", exc_info=True)
+        return set()
+
+
+def _apply_always_on(skills: list[Skill]) -> list[Skill]:
+    """Stamp ``Skill.always_on`` from the DB-backed 常驻 set (in place)."""
+    try:
+        always_on = _load_always_on_names()
+        if always_on:
+            for skill in skills:
+                skill.always_on = skill.name in always_on
+    except Exception:
+        logger.debug("Failed to apply always_on flags to skills", exc_info=True)
+    return skills
+
+
 def _load_enabled_skills_sync() -> list[Skill]:
-    return _drop_archived_skills(list(get_or_new_skill_storage().load_skills(enabled_only=True)))
+    return _apply_always_on(_drop_archived_skills(list(get_or_new_skill_storage().load_skills(enabled_only=True))))
 
 
 def _start_enabled_skills_refresh_thread() -> None:
@@ -188,7 +238,7 @@ def _get_enabled_skills_for_config(app_config: AppConfig | None = None) -> list[
     """
     if app_config is None:
         return _get_enabled_skills()
-    return _drop_archived_skills(list(get_or_new_skill_storage(app_config=app_config).load_skills(enabled_only=True)))
+    return _apply_always_on(_drop_archived_skills(list(get_or_new_skill_storage(app_config=app_config).load_skills(enabled_only=True))))
 
 
 def _skill_mutability_label(category: SkillCategory | str) -> str:
diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/agents/middlewares/summarization_middleware.py b/offline-backend-20260512/backend/packages/harness/deerflow/agents/middlewares/summarization_middleware.py
index c0a7b3e6..969a9910 100644
--- a/offline-backend-20260512/backend/packages/harness/deerflow/agents/middlewares/summarization_middleware.py
+++ b/offline-backend-20260512/backend/packages/harness/deerflow/agents/middlewares/summarization_middleware.py
@@ -314,16 +314,36 @@ class DeerFlowSummarizationMiddleware(SummarizationMiddleware):
             logger.debug("summarization: thread_id unresolved; cross-turn summary cache disabled for this call")
         cached = self._get_cache_entry(thread_id) if thread_id else None
         present_ids = {m.id for m in messages}
-        if cached is not None and cached.reusable and cached.summarized_ids <= present_ids:
+        cache_usable = cached is not None and cached.reusable and cached.summarized_ids <= present_ids
+        if cache_usable:
             preserved = [m for m in messages if m.id not in cached.summarized_ids]
             effective = [self._summary_message(cached.summary_text), *preserved]
-            if not self._should_summarize(effective, self.token_counter(effective)):
+            # Decide re-summarization on the *counted* size of the compressed
+            # context only — NOT ``_should_summarize`` — because that folds in the
+            # last AIMessage's reported ``usage_metadata`` (system prompt + tool
+            # schemas, fixed overhead compression can't shrink). Using it here
+            # would re-summarize on every single turn once that overhead alone
+            # approaches the trigger.
+            if not self._exceeds_trigger_by_count(effective):
                 # Refresh LRU recency and reuse without an LLM call.
                 self._touch_cache_entry(thread_id)
                 return ("reuse", effective)
             # New messages have pushed the compressed context back over the
             # threshold — fall through and re-summarize a larger prefix.
 
+        # Genuine (blocking) LLM summarize path. Log *why* we are paying for it so
+        # an "it compacts every turn" report is diagnosable straight from the
+        # backend log: cache=miss → thread_id/key problem; unusable → message-id
+        # instability; stale → preserved genuinely grew back over the trigger.
+        cache_state = "stale" if cache_usable else ("miss" if cached is None else "unusable")
+        logger.info(
+            "summarization: compacting thread=%s cache=%s counted_tokens=%d msgs=%d",
+            thread_id,
+            cache_state,
+            self.token_counter(messages),
+            len(messages),
+        )
+
         cutoff_index = self._determine_cutoff_index(messages)
         if cutoff_index <= 0:
             return None
@@ -386,6 +406,25 @@ class DeerFlowSummarizationMiddleware(SummarizationMiddleware):
             while len(self._summary_cache) > _SUMMARY_CACHE_MAX_THREADS:
                 self._summary_cache.popitem(last=False)
 
+    def _exceeds_trigger_by_count(self, messages: list[AnyMessage]) -> bool:
+        """Trigger check using only the *counted* size of ``messages``.
+
+        Deliberately ignores the stock middleware's reported-``usage_metadata``
+        heuristic: that number includes fixed system-prompt + tool-schema
+        overhead which compression can never shrink, so relying on it to decide
+        *re*-summarization makes a long, tool-heavy thread re-compact on every
+        turn. The first-ever decision to compress still uses the full
+        ``_should_summarize`` (reported usage is a fair signal there); only the
+        cache-reuse-vs-recompact decision uses this stricter, content-only check.
+        """
+        total = self.token_counter(messages)
+        for kind, value in getattr(self, "_trigger_conditions", []):
+            if kind == "messages" and len(messages) >= value:
+                return True
+            if kind == "tokens" and total >= value:
+                return True
+        return False
+
     def _apply_stickiness_headroom(self, messages: list[AnyMessage], cutoff_index: int) -> int:
         """Advance ``cutoff_index`` (keep fewer recent messages) so the preserved
         suffix lands below every trigger threshold with headroom.
diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/client.py b/offline-backend-20260512/backend/packages/harness/deerflow/client.py
index 69686db2..35374e24 100644
--- a/offline-backend-20260512/backend/packages/harness/deerflow/client.py
+++ b/offline-backend-20260512/backend/packages/harness/deerflow/client.py
@@ -910,7 +910,6 @@ class DeerFlowClient:
         current_skill_config = extensions_config.skills.get(name)
         extensions_config.skills[name] = SkillStateConfig(
             enabled=enabled,
-            always_on=current_skill_config.always_on if current_skill_config else False,
             display=current_skill_config.display if current_skill_config else None,
         )
 
diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/config/extensions_config.py b/offline-backend-20260512/backend/packages/harness/deerflow/config/extensions_config.py
index 8b1308f8..0a0c9f1f 100644
--- a/offline-backend-20260512/backend/packages/harness/deerflow/config/extensions_config.py
+++ b/offline-backend-20260512/backend/packages/harness/deerflow/config/extensions_config.py
@@ -161,10 +161,6 @@ class SkillStateConfig(BaseModel):
     """Configuration for a single skill's state."""
 
     enabled: bool = Field(default=True, description="Whether this skill is enabled")
-    always_on: bool = Field(
-        default=False,
-        description="常驻(免压缩):开启技能压缩后,本技能仍以全文形式注入系统提示词,而非降级为索引+search_skills 检索",
-    )
     display: SkillDisplayConfig | None = Field(default=None, description="Optional frontend display adapter")
 
 
@@ -317,21 +313,6 @@ class ExtensionsConfig(BaseModel):
             return skill_category in ("public", "custom")
         return skill_config.enabled
 
-    def is_skill_always_on(self, skill_name: str) -> bool:
-        """Check if a skill is marked 常驻(免压缩).
-
-        Args:
-            skill_name: Name of the skill
-
-        Returns:
-            True if the skill should bypass skill compression, False otherwise
-            (default for skills without an explicit config entry).
-        """
-        skill_config = self.skills.get(skill_name)
-        if skill_config is None:
-            return False
-        return skill_config.always_on
-
 
 _extensions_config: ExtensionsConfig | None = None
 
diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/migrations/versions/20260616_02_skill_favorites.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/migrations/versions/20260616_02_skill_favorites.py
new file mode 100644
index 00000000..0a223a3b
--- /dev/null
+++ b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/migrations/versions/20260616_02_skill_favorites.py
@@ -0,0 +1,54 @@
+"""我的技能 (skill favorites) table.
+
+New table ``skill_favorites`` — the per-user "我的技能" collection, mirroring
+``agent_favorites``. A 岗位 (position) materializes its skill tag bindings into
+this table (``origin='position'``) so granted skills land in the member's 我的
+list while the browse-all 广场 stays complete.
+
+Note: the running app also auto-creates this table via ``create_all`` on
+startup (new tables are always created, even on existing DBs); this migration
+keeps the alembic history consistent per the repo convention (see
+backend/CLAUDE.md → Schema Migrations).
+
+Revision ID: 20260616_02
+Revises: 20260616_01
+Create Date: 2026-06-16
+"""
+
+from __future__ import annotations
+
+import sqlalchemy as sa
+from alembic import op
+from sqlalchemy import inspect
+
+
+revision = "20260616_02"
+down_revision = "20260616_01"
+branch_labels = None
+depends_on = None
+
+
+def _has_table(table_name: str) -> bool:
+    return table_name in inspect(op.get_bind()).get_table_names()
+
+
+def upgrade() -> None:
+    if not _has_table("skill_favorites"):
+        op.create_table(
+            "skill_favorites",
+            sa.Column("id", sa.String(64), primary_key=True),
+            sa.Column("user_id", sa.String(64), nullable=False),
+            sa.Column("skill_name", sa.String(64), nullable=False),
+            sa.Column("origin", sa.String(16), nullable=False, server_default="user"),
+            sa.Column("position_id", sa.String(64), nullable=True),
+            sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
+            sa.UniqueConstraint("user_id", "skill_name", name="uq_skill_favorites_user_skill"),
+        )
+        op.create_index("ix_skill_favorites_user_id", "skill_favorites", ["user_id"])
+        op.create_index("ix_skill_favorites_skill_name", "skill_favorites", ["skill_name"])
+        op.create_index("ix_skill_favorites_position_id", "skill_favorites", ["position_id"])
+
+
+def downgrade() -> None:
+    if _has_table("skill_favorites"):
+        op.drop_table("skill_favorites")
diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/migrations/versions/20260616_03_skill_always_on.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/migrations/versions/20260616_03_skill_always_on.py
new file mode 100644
index 00000000..68d458bd
--- /dev/null
+++ b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/migrations/versions/20260616_03_skill_always_on.py
@@ -0,0 +1,45 @@
+"""Add always_on (常驻/免压缩) column to skills table.
+
+Revision ID: 20260616_03
+Revises: 20260616_02
+Create Date: 2026-06-16
+"""
+
+from __future__ import annotations
+
+import sqlalchemy as sa
+from alembic import op
+from sqlalchemy import inspect
+
+
+revision = "20260616_03"
+down_revision = "20260616_02"
+branch_labels = None
+depends_on = None
+
+
+def _has_table(table_name: str) -> bool:
+    inspector = inspect(op.get_bind())
+    return table_name in inspector.get_table_names()
+
+
+def _has_column(table_name: str, column_name: str) -> bool:
+    if not _has_table(table_name):
+        return False
+    inspector = inspect(op.get_bind())
+    return any(col["name"] == column_name for col in inspector.get_columns(table_name))
+
+
+def upgrade() -> None:
+    # ``skills`` may not exist yet on a brand-new DB — create_all handles that
+    # case with the column already present, so only patch existing tables.
+    if _has_table("skills") and not _has_column("skills", "always_on"):
+        op.add_column(
+            "skills",
+            sa.Column("always_on", sa.Boolean(), nullable=False, server_default="0"),
+        )
+
+
+def downgrade() -> None:
+    if _has_column("skills", "always_on"):
+        op.drop_column("skills", "always_on")
diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/models/__init__.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/models/__init__.py
index 4db6bcfc..535487b0 100644
--- a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/models/__init__.py
+++ b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/models/__init__.py
@@ -60,7 +60,7 @@ from deerflow.persistence.scheduled_tasks.model import (
     ScheduledTaskSubscriptionRow,
     SchedulerThreadRow,
 )
-from deerflow.persistence.skills.model import SkillDisplayAdapterRow, SkillRow
+from deerflow.persistence.skills.model import SkillDisplayAdapterRow, SkillFavoriteRow, SkillRow
 from deerflow.persistence.tags.model import TagAssignmentRow, TagRow
 from deerflow.persistence.thread_meta.model import ThreadMetaRow
 from deerflow.persistence.embed_sessions.model import EmbedSessionRow
@@ -110,6 +110,7 @@ __all__ = [
     "SchedulerThreadRow",
     "SkillRow",
     "SkillDisplayAdapterRow",
+    "SkillFavoriteRow",
     "TagAssignmentRow",
     "TagRow",
     "EmbedSessionRow",
diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/skills/base.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/skills/base.py
index f84ea8a5..69390001 100644
--- a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/skills/base.py
+++ b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/skills/base.py
@@ -85,3 +85,23 @@ class SkillStore(ABC):
         DB round-trip per candidate skill.
         """
         raise NotImplementedError
+
+    @abstractmethod
+    async def list_always_on_names(self) -> set[str]:
+        """Return the set of skill names marked 常驻(免压缩) (``always_on = True``).
+
+        Batch query used by the prompt builder to decide which skills bypass
+        skill compression — one query instead of a per-skill round-trip.
+        """
+        raise NotImplementedError
+
+    @abstractmethod
+    async def list_favorite_skill_names(self, user_id: str) -> set[str]:
+        """Return the skill names in ``user_id``'s "我的技能" collection.
+
+        These are skills the user does not own but that landed in their list —
+        currently only via their 岗位 (position) sync (``origin='position'``).
+        Owned skills are NOT included here (the caller detects those via
+        ``owner_user_id``).
+        """
+        raise NotImplementedError
diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/skills/memory.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/skills/memory.py
index 04f29409..e5f1adb6 100644
--- a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/skills/memory.py
+++ b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/skills/memory.py
@@ -20,12 +20,15 @@ def _copy(record: dict[str, Any] | None) -> dict[str, Any] | None:
 def _with_derived(record: dict[str, Any]) -> dict[str, Any]:
     out = dict(record)
     out["builtin"] = out.get("owner_user_id") is None
+    out.setdefault("always_on", False)
     return out
 
 
 class MemorySkillStore(SkillStore):
     def __init__(self) -> None:
         self._skills: dict[str, dict[str, Any]] = {}
+        # user_id -> {skill_name}
+        self._favorites: dict[str, set[str]] = {}
 
     @staticmethod
     def _is_visible(record: dict[str, Any], user_id: str) -> bool:
@@ -71,6 +74,7 @@ class MemorySkillStore(SkillStore):
             "published": bool(data.get("published", False)),
             "detail": data.get("detail"),
             "name_zh": data.get("name_zh"),
+            "always_on": bool(data.get("always_on", False)),
             "square_id": str(data.get("square_id") or ""),
             "created_at": _now(),
             "updated_at": _now(),
@@ -107,6 +111,8 @@ class MemorySkillStore(SkillStore):
             record["detail"] = data["detail"]
         if "name_zh" in data:
             record["name_zh"] = data["name_zh"]
+        if "always_on" in data:
+            record["always_on"] = bool(data["always_on"])
         if "square_id" in data:
             record["square_id"] = str(data["square_id"] or "")
         record["updated_at"] = _now()
@@ -124,6 +130,8 @@ class MemorySkillStore(SkillStore):
             record["detail"] = data["detail"]
         if "name_zh" in data:
             record["name_zh"] = data["name_zh"]
+        if "always_on" in data:
+            record["always_on"] = bool(data["always_on"])
         if "square_id" in data:
             record["square_id"] = str(data["square_id"] or "")
         record["updated_at"] = _now()
@@ -153,3 +161,13 @@ class MemorySkillStore(SkillStore):
 
     async def list_published_names(self) -> set[str]:
         return {name for name, record in self._skills.items() if record.get("published")}
+
+    async def list_always_on_names(self) -> set[str]:
+        return {name for name, record in self._skills.items() if record.get("always_on")}
+
+    async def list_favorite_skill_names(self, user_id: str) -> set[str]:
+        return set(self._favorites.get(user_id, set()))
+
+    # test helper — not part of the abstract API
+    def add_favorite(self, user_id: str, skill_name: str) -> None:
+        self._favorites.setdefault(user_id, set()).add(skill_name)
diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/skills/model.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/skills/model.py
index d631f112..50827767 100644
--- a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/skills/model.py
+++ b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/skills/model.py
@@ -10,10 +10,11 @@ from __future__ import annotations
 
 from datetime import UTC, datetime
 
-from sqlalchemy import Boolean, DateTime, String, Text
+from sqlalchemy import Boolean, DateTime, String, Text, UniqueConstraint
 from sqlalchemy.orm import Mapped, mapped_column
 
 from deerflow.persistence.base import Base
+from deerflow.persistence.types import BeijingDateTime
 
 
 class SkillRow(Base):
@@ -33,6 +34,10 @@ class SkillRow(Base):
     # the on-disk SKILL.md) so it can be set for built-in skills too, and so the
     # shared source file is never rewritten.
     name_zh: Mapped[str | None] = mapped_column(String(128), nullable=True, default=None)
+    # 常驻(免压缩):开启技能压缩后,标记此项的技能仍以全文形式注入系统提示词,而非
+    # 降级为索引 + search_skills 检索。存 DB(而非 extensions_config.json),内置技能
+    # 也能标记,且避免在某些部署里回写源码目录下的配置文件导致报错。
+    always_on: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False, server_default="0", index=True)
     # Publish square (发布广场) this skill is filed under once published. Pure
     # organizational category, no access control. Empty string = unassigned →
     # the listing layer resolves it to the default square. One square per skill.
@@ -41,6 +46,32 @@ class SkillRow(Base):
     updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False, default=lambda: datetime.now(UTC), onupdate=lambda: datetime.now(UTC))
 
 
+class SkillFavoriteRow(Base):
+    """A user's "我的技能" membership for a skill they don't own.
+
+    Mirrors :class:`deerflow.persistence.agents.model.AgentFavoriteRow`: lets a
+    skill land in a user's "我的" list without the user owning it. Skills are
+    keyed by ``name`` (the on-disk folder / DB primary key), so this row
+    references ``skill_name`` rather than an opaque id. One row per
+    (user, skill); the unique constraint makes the grant idempotent.
+    """
+
+    __tablename__ = "skill_favorites"
+    __table_args__ = (UniqueConstraint("user_id", "skill_name", name="uq_skill_favorites_user_skill"),)
+
+    id: Mapped[str] = mapped_column(String(64), primary_key=True)
+    user_id: Mapped[str] = mapped_column(String(64), nullable=False, index=True)
+    skill_name: Mapped[str] = mapped_column(String(64), nullable=False, index=True)
+    # ``"user"`` = the user added it themselves (reserved for a future favorite
+    # button); ``"position"`` = auto-granted by the user's 岗位 (position) via
+    # the subscription sync engine. Sync only ever adds/removes ``"position"``
+    # rows, so a user's own picks are never clobbered.
+    origin: Mapped[str] = mapped_column(String(16), nullable=False, default="user", server_default="user")
+    # The position that granted this favorite (only set when ``origin='position'``).
+    position_id: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True)
+    created_at: Mapped[datetime] = mapped_column(BeijingDateTime(), nullable=False, default=lambda: datetime.now(UTC))
+
+
 class SkillDisplayAdapterRow(Base):
     __tablename__ = "skill_display_adapters"
 
diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/skills/sql.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/skills/sql.py
index 30c1d90c..78004c07 100644
--- a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/skills/sql.py
+++ b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/skills/sql.py
@@ -9,7 +9,7 @@ from sqlalchemy import func, or_, select
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
 
 from deerflow.persistence.skills.base import SkillStore
-from deerflow.persistence.skills.model import SkillRow
+from deerflow.persistence.skills.model import SkillFavoriteRow, SkillRow
 
 
 def _row_to_dict(row: SkillRow) -> dict[str, Any]:
@@ -65,6 +65,7 @@ class SkillRepository(SkillStore):
             published=bool(data.get("published", False)),
             detail=data.get("detail"),
             name_zh=data.get("name_zh"),
+            always_on=bool(data.get("always_on", False)),
             square_id=str(data.get("square_id") or ""),
             created_at=now,
             updated_at=now,
@@ -110,6 +111,8 @@ class SkillRepository(SkillStore):
                 row.detail = data["detail"]
             if "name_zh" in data:
                 row.name_zh = data["name_zh"]
+            if "always_on" in data:
+                row.always_on = bool(data["always_on"])
             if "square_id" in data:
                 row.square_id = str(data["square_id"] or "")
             row.updated_at = datetime.now(UTC)
@@ -130,6 +133,8 @@ class SkillRepository(SkillStore):
                 row.detail = data["detail"]
             if "name_zh" in data:
                 row.name_zh = data["name_zh"]
+            if "always_on" in data:
+                row.always_on = bool(data["always_on"])
             if "square_id" in data:
                 row.square_id = str(data["square_id"] or "")
             row.updated_at = datetime.now(UTC)
@@ -171,3 +176,15 @@ class SkillRepository(SkillStore):
         async with self._sf() as session:
             result = await session.execute(stmt)
             return {row[0] for row in result.all()}
+
+    async def list_always_on_names(self) -> set[str]:
+        stmt = select(SkillRow.name).where(SkillRow.always_on.is_(True))
+        async with self._sf() as session:
+            result = await session.execute(stmt)
+            return {row[0] for row in result.all()}
+
+    async def list_favorite_skill_names(self, user_id: str) -> set[str]:
+        stmt = select(SkillFavoriteRow.skill_name).where(SkillFavoriteRow.user_id == user_id)
+        async with self._sf() as session:
+            result = await session.execute(stmt)
+            return {row[0] for row in result.all()}
diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/services/position_sync.py b/offline-backend-20260512/backend/packages/harness/deerflow/services/position_sync.py
index eda5bb47..bc3ee1e1 100644
--- a/offline-backend-20260512/backend/packages/harness/deerflow/services/position_sync.py
+++ b/offline-backend-20260512/backend/packages/harness/deerflow/services/position_sync.py
@@ -1,9 +1,9 @@
 """岗位 (position) → resource subscription **sync engine**.
 
-A user's position decides which agents / scheduled tasks / custom prompts they
-are subscribed to, and which skills / recommended questions they can see. This
-service turns a position's *tag bindings* into concrete, materialized rows in
-the existing "我的" tables (idempotent, re-runnable).
+A user's position decides which agents / skills / scheduled tasks they are
+subscribed to, and which recommended questions they can see. This service turns
+a position's *tag bindings* into concrete, materialized rows in the existing
+"我的" tables (idempotent, re-runnable).
 
 Resolution rule (per resource type):
 
@@ -14,15 +14,19 @@ Resolution rule (per resource type):
 * For a resource type **without** bound tags → the *default set*:
   - ``agent``               → all built-in agents (``user_id IS NULL``)
   - ``scheduled_task``      → empty (no built-in tasks)
-  - ``skill`` / ``recommended_question`` → *unrestricted* (handled at query
-    time, never materialized — see :meth:`resolve_visible`)
+  - ``skill``               → empty (the 广场 still shows every visible skill;
+    a position only *adds* its tagged skills to the member's "我的技能")
+  - ``recommended_question`` → *unrestricted* (handled at query time, never
+    materialized — see :meth:`resolve_visible`)
 
 Materialization (``sync_user``) only ever touches rows it created itself
 (``origin='position'``), so a user's own favorites / subscriptions are never
-clobbered. The skill + recommended-question visibility filters are computed live
-by :meth:`resolve_visible` and are *not* materialized. (The admin-configured
-prompts shown to users ARE the recommended questions — there is no separate
-prompt-template store.)
+clobbered. ``skill`` is materialized into ``skill_favorites`` exactly like
+``agent`` → ``agent_favorites``: the tagged skills are *added* to the user's
+"我的技能" while the browse-all 广场 stays complete (it is **not** restricted).
+Only recommended-question visibility is computed live by :meth:`resolve_visible`
+and is *not* materialized. (The admin-configured prompts shown to users ARE the
+recommended questions — there is no separate prompt-template store.)
 """
 
 from __future__ import annotations
@@ -37,14 +41,15 @@ from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
 from deerflow.persistence.agents.model import AgentFavoriteRow, AgentRow
 from deerflow.persistence.positions.base import PositionStore
 from deerflow.persistence.scheduled_tasks.model import ScheduledTaskSubscriptionRow
+from deerflow.persistence.skills.model import SkillFavoriteRow
 from deerflow.persistence.tags.base import TagStore
 
 logger = logging.getLogger(__name__)
 
 # Materialized into "我的" tables by sync_user.
-_MATERIALIZED_TYPES = ("agent", "scheduled_task")
+_MATERIALIZED_TYPES = ("agent", "skill", "scheduled_task")
 # Resolved live at query time (visibility filters), never materialized.
-_VISIBILITY_TYPES = ("skill", "recommended_question")
+_VISIBILITY_TYPES = ("recommended_question",)
 
 
 class PositionSyncService:
@@ -64,9 +69,10 @@ class PositionSyncService:
     async def resolve_visible(self, user_id: str, resource_type: str) -> set[str] | None:
         """Return the set of visible ``resource_type`` ids, or ``None`` = unrestricted.
 
-        Used by the skills list + recommended-questions filters. ``None`` means
-        "no position restriction → show everything"; a set means "restrict to
-        exactly these ids".
+        Used by the recommended-questions filter (the only remaining query-time
+        visibility type; skills are now *materialized*, not filtered). ``None``
+        means "no position restriction → show everything"; a set means "restrict
+        to exactly these ids".
         """
         pid = await self._effective_position_id(user_id)
         if not pid:
@@ -101,6 +107,7 @@ class PositionSyncService:
         pid = await self._effective_position_id(user_id)
         async with self._sf() as session:
             await self._sync_agents(session, user_id, pid)
+            await self._sync_skills(session, user_id, pid)
             await self._sync_tasks(session, user_id, pid)
             await session.commit()
 
@@ -116,6 +123,18 @@ class PositionSyncService:
             if agent_id not in target:
                 await session.delete(row)
 
+    async def _sync_skills(self, session: AsyncSession, user_id: str, position_id: str | None) -> None:
+        target = await self._resolve_materialized(user_id, "skill", session)
+        rows = list((await session.execute(select(SkillFavoriteRow).where(SkillFavoriteRow.user_id == user_id))).scalars())
+        owned_any = {r.skill_name for r in rows}
+        position_rows = {r.skill_name: r for r in rows if r.origin == "position"}
+        now = datetime.now(UTC)
+        for skill_name in target - owned_any:
+            session.add(SkillFavoriteRow(id=uuid.uuid4().hex, user_id=user_id, skill_name=skill_name, origin="position", position_id=position_id, created_at=now))
+        for skill_name, row in position_rows.items():
+            if skill_name not in target:
+                await session.delete(row)
+
     async def _sync_tasks(self, session: AsyncSession, user_id: str, position_id: str | None) -> None:
         target = await self._resolve_materialized(user_id, "scheduled_task", session)
         rows = list((await session.execute(select(ScheduledTaskSubscriptionRow).where(ScheduledTaskSubscriptionRow.user_id == user_id))).scalars())
diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/skills/storage/skill_storage.py b/offline-backend-20260512/backend/packages/harness/deerflow/skills/storage/skill_storage.py
index 77d10e0b..6f03eb30 100644
--- a/offline-backend-20260512/backend/packages/harness/deerflow/skills/storage/skill_storage.py
+++ b/offline-backend-20260512/backend/packages/harness/deerflow/skills/storage/skill_storage.py
@@ -247,7 +247,6 @@ class SkillStorage(ABC):
             extensions_config = ExtensionsConfig.from_file()
             for skill in skills:
                 skill.enabled = extensions_config.is_skill_enabled(skill.name, skill.category)
-                skill.always_on = extensions_config.is_skill_always_on(skill.name)
         except Exception as e:
             logger.warning("Failed to load extensions config: %s", e)
 
diff --git a/offline-backend-20260512/backend/scripts/sql/column_additions_mysql.sql b/offline-backend-20260512/backend/scripts/sql/column_additions_mysql.sql
index ec63fc13..a0465908 100644
--- a/offline-backend-20260512/backend/scripts/sql/column_additions_mysql.sql
+++ b/offline-backend-20260512/backend/scripts/sql/column_additions_mysql.sql
@@ -65,6 +65,9 @@ CALL _add_column_if_missing('skills', 'detail', 'TEXT NULL DEFAULT NULL');
 -- square_id: 技能所属的发布广场(发布广场功能),对应 migration 20260611_01。
 --          ORM: SkillRow.square_id = String(64), nullable=False, default=''。
 CALL _add_column_if_missing('skills', 'square_id', "VARCHAR(64) NOT NULL DEFAULT ''");
+-- always_on: 常驻(免压缩) —— 开启技能压缩后仍全文注入系统提示词,对应 migration 20260616_03。
+--          ORM: SkillRow.always_on = Boolean, nullable=False, default=False。
+CALL _add_column_if_missing('skills', 'always_on', 'TINYINT(1) NOT NULL DEFAULT 0');
 
 -- ---- agents ------------------------------------------------------------------
 -- square_id: 智能体所属的发布广场(发布广场功能),对应 migration 20260611_01。
@@ -133,7 +136,7 @@ DEALLOCATE PREPARE stmt;
 SELECT table_name, column_name
   FROM information_schema.columns
  WHERE table_schema = DATABASE()
-   AND ((table_name = 'skills' AND column_name IN ('name_zh', 'detail', 'square_id'))
+   AND ((table_name = 'skills' AND column_name IN ('name_zh', 'detail', 'square_id', 'always_on'))
         OR (table_name = 'agents' AND column_name IN ('square_id'))
         OR (table_name = 'roundtable_drafts' AND column_name = 'task_id'))
  ORDER BY table_name, column_name;