Skills

PROMPTS_DIR = Path(__file__).parent.parent / "prompts"
SKILLS_DIR = Path(__file__).parent.parent / "skills"
 
#: Mount point the agent sees. `SkillsMiddleware` scans one level below it.
SKILLS_ROOT = "/skills/"
 
SKILLS_NAMESPACE = ("cockpit", "deep-agents-skills")
 
#: Custom stream event name the Angular panel listens for.
SKILLS_EVENT = "deep_agents.skills"

def _seed_skills_store() -> InMemoryStore:
    """Load the bundled skill folders into a process-local store.
 
    Read once at import through a writable backend. Runtime access is mounted
    through `ReadOnlySkillsBackend`, which denies every mutation entry point.
    """
    store = InMemoryStore()
    backend = StoreBackend(namespace=lambda _runtime: SKILLS_NAMESPACE, store=store)
    for path in sorted(SKILLS_DIR.rglob("*.md")):
        # Paths are stored WITHOUT the mount prefix. `CompositeBackend` strips
        # the route prefix before delegating, so a store seeded at
        # `/skills/runway-analysis/...` would surface as
        # `/skills/skills/runway-analysis/...` to the agent.
        backend.write(f"/{path.relative_to(SKILLS_DIR).as_posix()}", path.read_text())
    return store
 
 
SKILLS_STORE = _seed_skills_store()

"""Read-only StoreBackend mount for bundled Skills on the shared deployment."""
 
from deepagents.backends import StoreBackend
from deepagents.backends.protocol import (
    PERMISSION_DENIED,
    DeleteResult,
    EditResult,
    FileUploadResponse,
    WriteResult,
)
 
 
class ReadOnlySkillsBackend(StoreBackend):
    """Preserve StoreBackend reads while denying every mutation entry point.
 
    Seed the store through a regular StoreBackend before mounting this class.
    The async methods are overridden too: StoreBackend's async implementations
    access the store directly rather than delegating to these sync methods.
    """
 
    def write(self, file_path: str, content: str) -> WriteResult:
        return WriteResult(error=PERMISSION_DENIED)
 
    async def awrite(self, file_path: str, content: str) -> WriteResult:
        return self.write(file_path, content)
 
    def edit(
        self, file_path: str, old_string: str, new_string: str, replace_all: bool = False
    ) -> EditResult:
        return EditResult(error=PERMISSION_DENIED)
 
    async def aedit(
        self, file_path: str, old_string: str, new_string: str, replace_all: bool = False
    ) -> EditResult:
        return self.edit(file_path, old_string, new_string, replace_all)
 
    def delete(self, file_path: str) -> DeleteResult:
        return DeleteResult(error=PERMISSION_DENIED)
 
    async def adelete(self, file_path: str) -> DeleteResult:
        return self.delete(file_path)
 
    def upload_files(self, files: list[tuple[str, bytes]]) -> list[FileUploadResponse]:
        return [FileUploadResponse(path=path, error=PERMISSION_DENIED) for path, _ in files]
 
    async def aupload_files(self, files: list[tuple[str, bytes]]) -> list[FileUploadResponse]:
        return self.upload_files(files)

def build_skills_agent():
    """Build the skills agent.
 
    `CompositeBackend` routes by path prefix, longest first. `/skills/` resolves
    to the seeded store; everything else falls through to `StateBackend`, so any
    notes the agent writes stay on the thread and never touch the skill mount.
    """
    return create_deep_agent(
        model=ChatOpenAI(model="gpt-4.1", temperature=0),
        tools=[lookup_field_elevation, lookup_runway_length, lookup_weather],
        system_prompt=(PROMPTS_DIR / "skills.md").read_text(),
        backend=CompositeBackend(
            default=StateBackend(),
            routes={
                SKILLS_ROOT: ReadOnlySkillsBackend(
                    namespace=lambda _runtime: SKILLS_NAMESPACE,
                    store=SKILLS_STORE,
                ),
            },
        ),
        skills=[SKILLS_ROOT],
        middleware=[SkillsVisibilityMiddleware()],
    )
 

class SkillsVisibilityMiddleware(AgentMiddleware):
    """Republish `skills_metadata` as a `custom` stream event.
 
    Skills keys are excluded from declared input/output schemas. Local compiled
    root values expose metadata; observed deployed API values omit private keys.
    Saved checkpoints remain authoritative. Optional load errors can appear when
    produced and can be omitted on normal loads. This live index event includes
    both keys, with an empty-list fallback for errors; it does not prove a
    confirmed read and makes no claim about Memory visibility.
    """
 
    @property
    def name(self) -> str:
        return "SkillsVisibilityMiddleware"
 
    def _emit(self, state: dict[str, Any]) -> None:
        metadata = state.get("skills_metadata")
        if metadata is None:
            return
        try:
            writer = get_stream_writer()
        except (RuntimeError, KeyError):
            return
        writer(
            {
                "name": SKILLS_EVENT,
                "data": {
                    "skills_metadata": metadata,
                    "skills_load_errors": state.get("skills_load_errors") or [],
                },
            }
        )
 
    def before_model(self, state: dict[str, Any], runtime: Any) -> None:  # noqa: ANN401, ARG002
        self._emit(state)
        return None
 
    def after_agent(self, state: dict[str, Any], runtime: Any) -> None:  # noqa: ANN401, ARG002
        self._emit(state)
        return None
 

import { injectCockpitRuntimeConnection } from '@threadplane/cockpit-telemetry';
import { ApplicationConfig } from '@angular/core';
import { provideAgent } from '@threadplane/langgraph';
 
export const appConfig: ApplicationConfig = {
  providers: [
    provideAgent(() => {
      const connection = injectCockpitRuntimeConnection();
      if (connection.adapter !== 'langgraph') {
        throw new Error('incompatible runtime');
      }
      return {
        apiUrl: connection.apiUrl,
        assistantId: connection.assistantId,
        clientOptions: connection.clientOptions,
      };
    }),
  ],
};

private readonly liveSkills = computed<Record<string, unknown>[] | null>(() => {
  for (const event of [...this.agent.customEvents()].reverse()) {
    if (event.name !== SKILLS_EVENT) continue;
    const metadata = (event.data as { skills_metadata?: unknown } | undefined)?.[
      'skills_metadata'
    ];
    if (Array.isArray(metadata) && metadata.length > 0) {
      return metadata as Record<string, unknown>[];
    }
  }
  return null;
});
 
private readonly settledSkills = computed<Record<string, unknown>[] | null>(() => {
  const metadata = (this.agent.value() as Record<string, unknown> | undefined)?.[
    'skills_metadata'
  ];
  return Array.isArray(metadata) && metadata.length > 0
    ? (metadata as Record<string, unknown>[])
    : null;
});
 
protected readonly skillsSource = computed<'live' | 'checkpoint' | 'none'>(() => {
  if (this.liveSkills()) return 'live';
  return this.settledSkills() ? 'checkpoint' : 'none';
});

/** Requested read_file paths, without validating completion or returned content. */
private readonly openedPaths = computed<string[]>(() => {
  const paths: string[] = [];
  for (const call of this.agent.toolCalls()) {
    if (call.name !== 'read_file') continue;
    const path = (call.args as Record<string, unknown> | undefined)?.['file_path'];
    if (typeof path === 'string' && !paths.includes(path)) paths.push(path);
  }
  return paths;
});

protected readonly skills = computed<SkillEntry[]>(() => {
  const source = this.liveSkills() ?? this.settledSkills() ?? [];
  const opened = this.openedPaths();
  return source.map((entry) => {
    const path = String(entry['path'] ?? '');
    const root = path.slice(0, path.lastIndexOf('/') + 1);
    return {
      name: String(entry['name'] ?? ''),
      description: String(entry['description'] ?? ''),
      path,
      root,
      opened: root ? opened.filter((candidate) => candidate.startsWith(root)) : [],
    };
  });
});

<div sidebar class="panel">
  <h3 class="cap">Skill Index</h3>
  <p class="source" data-testid="skills-source" [attr.data-source]="skillsSource()">
    {{ skillsSource() === 'live' ? 'streamed live' : skillsSource() === 'checkpoint' ? 'from checkpoint' : 'no source yet' }}
  </p>
  @if (skills().length === 0) {
    <p class="empty">No skills loaded</p>
  }
  @for (skill of skills(); track skill.name) {
    <div
      class="skill"
      data-testid="skill"
      [attr.data-name]="skill.name"
      [attr.data-opened]="skill.opened.length > 0 ? 'true' : 'false'"
    >
      <span class="skill__name">{{ skill.name }}</span>
      <p class="skill__description">{{ skill.description }}</p>
      @if (skill.opened.length === 0) {
        <p class="skill__files skill__files--idle">body not read</p>
      } @else {
        @for (file of skill.opened; track file) {
          <p class="skill__files" data-testid="skill-open">{{ file }}</p>
        }
      }
    </div>
  }
  <p class="hint">
    Only the name and description above were put in the model's prompt. Everything else was
    read on demand.
  </p>
</div>