feat(api): add scope-guarded profile write endpoints

docs/api.md

@@ -60,6 +60,50 @@ See [AG-UI](agui.md) for enablement, event shapes, and privacy boundaries.
   client can render create and edit forms from the server's definition instead
   of duplicating the field list.
 - `POST /agents/profiles/install` installs a profile.
+- `POST /agents/profiles` creates a profile in the local store from a supplied
+  document. Named distinctly from `install`, which takes a bare profile name or
+  an https:// URL rather than the document itself. The request carries `name`
+  and `content`; the two identities of a profile, its storage key and its
+  frontmatter `name`, must agree, so a mismatch is a 400 rather than a silent
+  rename. A conflicting name returns 409. Requires `cao:write` or `cao:admin`.
+- `PUT /agents/profiles/{name}` replaces an existing local-store profile and
+  never creates one. A request naming a built-in or provider-managed profile
+  returns 404 rather than writing a local file that would shadow the original.
+  Requires `cao:write` or `cao:admin`.
+- `DELETE /agents/profiles/{name}` removes a profile from the local store.
+  Requires `cao:write` or `cao:admin`, the same guard as create and replace, so
+  one credential covers the whole create/edit/delete cycle. Scopes are a flat
+  set rather than a hierarchy, so requiring admin here would 403 a caller
+  holding exactly `cao:write`. Built-ins are not deletable, for the same reason
+  they are not replaceable.
+- Both write routes run the profile validator on the exact submitted document
+  before persisting anything, so an invalid profile never reaches disk. Errors
+  reject the request with 400 and the findings attached; warnings do not block
+  the write and are returned in the response so a client can surface them after
+  a successful save.
+- The validator rejects non-string mapping keys. A profile is written as YAML,
+  which allows any scalar as a key, but the format is described by JSON Schema,
+  where object keys are strings. Without this rule `mcpServers: {1: {...}}`
+  validates clean and persists, then fails to load, since the model requires
+  string keys. Note YAML also auto-types an unquoted date, so `2026-01-01:` is a
+  date key rather than a string; quote such keys.
+- Every 400 from the profile write and source routes uses one `detail` shape,
+  `{"message", "errors"}`, so a client never has to switch on the type of
+  `detail`. `errors` is empty for a failure that is not attributable to a field,
+  but the key is always present. This covers rejected names as well as schema
+  findings. 404 and 409 keep FastAPI's conventional bare-string `detail`, since
+  the status code already discriminates and there are no findings to attach.
+- `GET /agents/profiles/{name}/source` returns a profile's document exactly as
+  stored. Use this, not `GET /agents/profiles/{name}`, when the document is
+  going to be edited and written back: that route returns the *resolved*
+  profile, having applied `${VAR}` substitution from the managed environment
+  file to the raw text before parsing. Round-tripping a resolved document
+  through a write would persist substituted values into a plaintext profile.
+  Requires `cao:read`, `cao:write`, or `cao:admin`. The pre-existing profile
+  reads beside it are ungated and stay that way, since tightening a shipped
+  route could break an existing unauthenticated reader; this one is gated
+  because it returns the stored bytes verbatim from every configured store,
+  including documents that fail to parse.
 - Template validation and preview require the selected template to include a
   `schema.json` file.
 - `/agents/providers` reports provider availability.

src/cli_agent_orchestrator/api/main.py

@@ -15,7 +15,18 @@
 from dataclasses import asdict
 from datetime import datetime, timezone
 from pathlib import Path
-from typing import Annotated, Any, AsyncIterator, Dict, List, Literal, Optional, Tuple, cast
+from typing import (
+    Annotated,
+    Any,
+    AsyncIterator,
+    Dict,
+    List,
+    Literal,
+    Optional,
+    Sequence,
+    Tuple,
+    cast,
+)
 
 from fastapi import (
     BackgroundTasks,
@@ -951,6 +962,61 @@ class ProfileValidationResponse(BaseModel):
     messages: List[ProfileValidationMessage] = Field(default_factory=list)
 
 
+class ProfileCreateRequest(BaseModel):
+    """Request body for ``POST /agents/profiles``.
+
+    ``name`` is explicit rather than parsed out of ``content`` so the conflict
+    target is unambiguous even when the document is malformed. When the
+    frontmatter also declares a ``name`` the two must agree; see
+    ``_assert_frontmatter_name_matches``.
+    """
+
+    name: str = Field(description="Profile name, used as the local-store filename stem")
+    content: str = Field(
+        max_length=262_144,
+        description="Full profile markdown, including YAML frontmatter",
+    )
+
+
+class ProfileReplaceRequest(BaseModel):
+    """Request body for ``PUT /agents/profiles/{name}``.
+
+    No ``name`` field: the path parameter is authoritative. Frontmatter that
+    declares a different name is rejected rather than silently renaming.
+    """
+
+    content: str = Field(
+        max_length=262_144,
+        description="Full profile markdown, including YAML frontmatter",
+    )
+
+
+class ProfileWriteResponse(BaseModel):
+    """Outcome of a profile create or replace.
+
+    ``warnings`` carries advisory findings that did not block the write, so a
+    client can surface them after a successful save. Errors never reach here;
+    they reject the request with 400.
+    """
+
+    name: str
+    warnings: List[ProfileValidationMessage] = Field(default_factory=list)
+
+
+class ProfileSourceResponse(BaseModel):
+    """A profile's document exactly as stored, with placeholders intact.
+
+    Distinct from ``GET /agents/profiles/{name}``, which returns the *parsed and
+    resolved* profile. That response runs ``resolve_env_vars`` over the raw text
+    before parsing, so managed ``${VAR}`` placeholders come back as their
+    substituted values. Round-tripping that through a write would persist
+    resolved secrets into a plaintext profile, so an editor must read from here.
+    """
+
+    name: str
+    content: str
+
+
 class MemorySummary(BaseModel):
     """Memory list entry. Excludes file_path (absolute server filesystem path)."""
 
@@ -2184,9 +2250,108 @@ async def get_agent_profile_schema_endpoint() -> Dict:
     return load_profile_schema()
 
 
+def _profile_write_rejection(message: str, findings: Sequence[Any] = ()) -> HTTPException:
+    """Build the one 400 a profile write route may return.
+
+    Every 400 from the profile write and source routes carries this shape,
+    ``{"message", "errors"}``, so a client parses one thing rather than switching
+    on ``type(detail)``. ``errors`` is empty for a failure that is not
+    attributable to a field, but the key is always present so a caller can
+    iterate it unconditionally.
+
+    Deliberately covers the service-raised ``InvalidProfileNameError`` paths too,
+    not only schema findings. An unsafe name is a rejected input just like a
+    schema violation, and returning a bare string for one and a dict for the
+    other reintroduces exactly the type-switching this removes. The 404 and 409
+    mappings keep FastAPI's conventional bare-string ``detail``: the status code
+    already tells a client what happened and there are no findings to attach.
+    """
+    return HTTPException(
+        status_code=status.HTTP_400_BAD_REQUEST,
+        detail={
+            "message": message,
+            "errors": [
+                {"severity": f.severity, "message": f.message, "path": f.path} for f in findings
+            ],
+        },
+    )
+
+
+def _validate_profile_for_write(name: str, content: str) -> List[ProfileValidationMessage]:
+    """Validate a submitted profile document and enforce name identity.
+
+    Shared by ``POST /agents/profiles`` and ``PUT /agents/profiles/{name}`` so the
+    two cannot drift apart on either rule.
+
+    Runs the same validator the CLI and ``POST /agents/profiles/validate`` use, on
+    the exact document being persisted rather than on a client-side approximation
+    of it. Error-severity findings reject the write; warnings are returned so a
+    client can surface them after a successful save.
+
+    A profile has two identities: the storage key (its filename stem) and the
+    frontmatter ``name``. ``parse_agent_profile_text`` treats the stem only as a
+    fallback when frontmatter omits ``name``, so the two can diverge and nothing
+    reconciles them: ``name: foo`` in ``bar.md`` loads as ``foo`` while being
+    addressed as ``bar``. Requiring them to agree closes that without introducing
+    a rename operation, which has its own failure semantics.
+
+    Args:
+        name: The storage name, authoritative.
+        content: The full profile document.
+
+    Returns:
+        The warning-severity findings, if any.
+
+    Raises:
+        HTTPException: 400 if the document is unparseable, carries an
+            error-severity finding, or declares a conflicting ``name``. See
+            :func:`_profile_write_rejection` for the shared ``detail`` shape.
+    """
+    import frontmatter
+
+    from cli_agent_orchestrator.services.profile_validator import validate_frontmatter
+
+    def _reject(message: str, findings: Sequence[Any] = ()) -> None:
+        raise _profile_write_rejection(message, findings)
+
+    # Parsed once here, then handed to validate_frontmatter as metadata.
+    # validate_profile_text would parse it again: its docstring exists precisely
+    # to keep callers from duplicating the parse, and this function needs the
+    # metadata anyway for the name check below.
+    try:
+        parsed = frontmatter.loads(content)
+    except Exception as exc:
+        _reject(f"Profile could not be parsed and was not written: {exc}")
+
+    findings = validate_frontmatter(parsed.metadata)
+
+    errors = [f for f in findings if f.severity == "error"]
+    if errors:
+        _reject("Profile failed validation and was not written.", errors)
+
+    declared = parsed.metadata.get("name")
+    if isinstance(declared, str) and declared != name:
+        _reject(
+            f"Frontmatter name '{declared}' does not match the profile name "
+            f"'{name}'. They must agree; renaming a profile is not supported "
+            f"through this endpoint."
+        )
+
+    return [
+        ProfileValidationMessage(severity=f.severity, message=f.message, path=f.path)
+        for f in findings
+        if f.severity == "warning"
+    ]
+
+
 @app.get("/agents/profiles/{name}")
 async def get_agent_profile_endpoint(name: str) -> Dict:
-    """Return the full parsed content of a named agent profile."""
+    """Return the full parsed content of a named agent profile.
+
+    Note this response is *resolved*: ``load_agent_profile`` applies
+    ``resolve_env_vars`` before parsing. Use ``GET /agents/profiles/{name}/source``
+    when the document is going to be edited and written back.
+    """
     try:
         profile = load_agent_profile(name)
         return profile.model_dump(exclude_none=True)
@@ -2222,6 +2387,148 @@ async def install_agent_profile_endpoint(
     return result
 
 
+@app.post("/agents/profiles", status_code=status.HTTP_201_CREATED)
+async def create_agent_profile_endpoint(
+    request: ProfileCreateRequest,
+    _scopes: List[str] = Depends(require_any_scope(SCOPE_WRITE, SCOPE_ADMIN)),
+) -> ProfileWriteResponse:
+    """Create a profile in the local store from a supplied document.
+
+    Named distinctly from ``POST /agents/profiles/install``, which installs from a
+    bare name or an https:// URL. This one takes the document itself in the body.
+
+    Validation runs on the exact submitted content before anything is persisted,
+    so an invalid profile never reaches disk. Conflict detection is delegated to
+    ``write_profile(overwrite=False)``, which checks for an existing file inside
+    the write lock; a pre-check here would sit outside that critical section and
+    let two concurrent creators both succeed.
+    """
+    from cli_agent_orchestrator.services.profile_store import (
+        InvalidProfileNameError,
+        ProfileExistsError,
+        write_profile,
+    )
+
+    warnings = _validate_profile_for_write(request.name, request.content)
+
+    try:
+        write_profile(request.name, request.content, overwrite=False)
+    except InvalidProfileNameError as exc:
+        raise _profile_write_rejection(str(exc))
+    except ProfileExistsError as exc:
+        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=str(exc))
+
+    return ProfileWriteResponse(name=request.name, warnings=warnings)
+
+
+@app.put("/agents/profiles/{name}")
+async def replace_agent_profile_endpoint(
+    name: str,
+    request: ProfileReplaceRequest,
+    _scopes: List[str] = Depends(require_any_scope(SCOPE_WRITE, SCOPE_ADMIN)),
+) -> ProfileWriteResponse:
+    """Replace an existing local-store profile. Never creates one.
+
+    Backed by ``replace_profile``, which requires the target to exist *inside* the
+    write lock. That is what makes a PUT naming a built-in or provider-managed
+    profile a 404 rather than a silent create: the local store is the only place
+    this resolves, so a built-in's name is simply not there. An upsert would
+    instead write a local file that shadows the built-in on load, manufacturing
+    exactly the condition ``duplicated_in`` exists to report.
+    """
+    from cli_agent_orchestrator.services.profile_store import (
+        InvalidProfileNameError,
+        ProfileNotFoundError,
+        replace_profile,
+    )
+
+    warnings = _validate_profile_for_write(name, request.content)
+
+    try:
+        replace_profile(name, request.content)
+    except InvalidProfileNameError as exc:
+        raise _profile_write_rejection(str(exc))
+    except ProfileNotFoundError as exc:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc))
+
+    return ProfileWriteResponse(name=name, warnings=warnings)
+
+
+@app.delete("/agents/profiles/{name}", status_code=status.HTTP_204_NO_CONTENT)
+async def delete_agent_profile_endpoint(
+    name: str,
+    _scopes: List[str] = Depends(require_any_scope(SCOPE_WRITE, SCOPE_ADMIN)),
+) -> None:
+    """Delete a profile from the local store.
+
+    Write-or-admin, the same guard as POST and PUT, so one credential completes
+    the whole create/edit/delete cycle that issue #510 specifies. Scopes are a
+    flat set here, not a hierarchy: ``require_any_scope`` tests membership, so
+    admin-only would 403 a caller holding exactly ``cao:write`` and leave a
+    client that can create and edit a profile unable to remove it.
+
+    Most other ``DELETE`` routes on this service do require admin alone, but they
+    remove *running or generated* state: sessions, terminals, workflows, flows,
+    and bulk memory. A profile is an authored document, closer to
+    ``DELETE /memory/relationships/{id}``, which is also write-or-admin. Removing
+    one stops no in-flight work and destroys nothing that cannot be re-authored,
+    and the deletion is already gated behind a confirmation in the UI.
+
+    Built-in and provider-managed profiles are not deletable for the same reason
+    they are not replaceable: ``delete_profile`` resolves only inside the local
+    store, so their names raise ``ProfileNotFoundError``.
+    """
+    from cli_agent_orchestrator.services.profile_store import (
+        InvalidProfileNameError,
+        ProfileNotFoundError,
+        delete_profile,
+    )
+
+    try:
+        delete_profile(name)
+    except InvalidProfileNameError as exc:
+        raise _profile_write_rejection(str(exc))
+    except ProfileNotFoundError as exc:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc))
+
+
+@app.get("/agents/profiles/{name}/source")
+async def get_agent_profile_source_endpoint(
+    name: str,
+    _scopes: List[str] = Depends(require_any_scope(SCOPE_READ, SCOPE_WRITE, SCOPE_ADMIN)),
+) -> ProfileSourceResponse:
+    """Return a profile's document exactly as stored, unresolved.
+
+    The authoring counterpart to ``GET /agents/profiles/{name}``. That route calls
+    ``load_agent_profile``, which applies ``resolve_env_vars`` to the raw text
+    *before* parsing, so substitution reaches the Markdown body as well as the
+    frontmatter, and the substitution source is the managed CAO ``.env`` file.
+    Using that response to pre-fill an editor and then PUT it back would persist
+    resolved secret values into a plaintext profile. ``safe_substitute`` leaves
+    unset variables intact, which would make the damage selective and silent.
+
+    Reads across all configured stores, not only the local one, so a built-in can
+    be fetched as the starting point for a clone. Writing it back still requires
+    the local store, which is enforced by the write routes.
+
+    Scope-gated, unlike the pre-existing profile reads beside it, following the
+    precedent set on #505: a newly added read route carries the gate, while
+    already-shipped ungated siblings are left alone because tightening them could
+    break an existing unauthenticated reader. Gating matters more here than on
+    those siblings because this route returns the stored bytes verbatim from the
+    local, provider, extra and built-in stores, including documents that fail to
+    parse, whereas the parsed route can only return what the model accepts.
+    """
+    from cli_agent_orchestrator.utils.agent_profiles import _read_agent_profile_source
+
+    try:
+        return ProfileSourceResponse(name=name, content=_read_agent_profile_source(name))
+    except FileNotFoundError as exc:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc))
+    except ValueError as exc:
+        raise _profile_write_rejection(str(exc))
+
+
 @app.get("/agents/providers")
 async def list_providers_endpoint() -> List[Dict]:
     """List available providers with installation status."""