repositories
loading repo index
repositories
loading repo index
repository
loading code, commits, and activity
Mirrored from https://github.com/yingqi-z20/Agent-libOS
stars
latest
clone command
git clone gitlawb://did:key:z6MkqRzA...RfoM/yingqi-z20-Agen...git clone gitlawb://did:key:z6MkqRzA.../yingqi-z20-Agen...d98dd2c9IPC1d ago| #1 | from __future__ import annotations |
| #2 | |
| #3 | from agent_libos.models.exceptions import ValidationError |
| #4 | from agent_libos.utils.ids import utc_now |
| #5 | from agent_libos.runtime.audit_manager import AuditManager |
| #6 | from agent_libos.skills.registry import RuntimeSkillRegistry |
| #7 | from agent_libos.skills.schema import SkillSpec |
| #8 | from agent_libos.skills.verifier import SkillVerifier |
| #9 | from agent_libos.storage import SQLiteStore |
| #10 | |
| #11 | |
| #12 | class SkillLinker: |
| #13 | def __init__( |
| #14 | self, |
| #15 | store: SQLiteStore, |
| #16 | registry: RuntimeSkillRegistry, |
| #17 | audit: AuditManager, |
| #18 | verifier: SkillVerifier | None = None, |
| #19 | ): |
| #20 | self.store = store |
| #21 | self.registry = registry |
| #22 | self.audit = audit |
| #23 | self.verifier = verifier or SkillVerifier() |
| #24 | |
| #25 | def dlopen_skill(self, pid: str, skill: SkillSpec, require_signature: bool = False) -> str: |
| #26 | ok, errors = self.verifier.verify(skill, require_signature=require_signature) |
| #27 | if not ok: |
| #28 | raise ValidationError("; ".join(errors)) |
| #29 | self.registry.register(skill) |
| #30 | process = self.store.get_process(pid) |
| #31 | if process is not None: |
| #32 | process.loaded_skills[skill.skill_id] = skill.version |
| #33 | process.updated_at = utc_now() |
| #34 | self.store.update_process(process) |
| #35 | self.audit.record( |
| #36 | actor=pid, |
| #37 | action="skill.load", |
| #38 | target=f"skill:{skill.skill_id}", |
| #39 | decision={"version": skill.version, "signed": skill.signed}, |
| #40 | ) |
| #41 | return skill.skill_id |
| #42 | |
| #43 | def unload_skill(self, pid: str, skill_id: str) -> None: |
| #44 | process = self.store.get_process(pid) |
| #45 | if process is not None: |
| #46 | process.loaded_skills.pop(skill_id, None) |
| #47 | process.updated_at = utc_now() |
| #48 | self.store.update_process(process) |
| #49 | self.audit.record(actor=pid, action="skill.unload", target=f"skill:{skill_id}") |
| #50 | |
| #51 |