oats.lsp

Language Server Protocol client integration.

oats.lsp.client

Minimal stdio LSP client/manager for coder.

This is intentionally lightweight: it targets the highest-value editor-style operations without pulling in additional heavy dependencies.

oats.lsp.client.path_to_uri(path)[source]

Convert a filesystem path to a file:// URI.

Return type:

str

oats.lsp.client.uri_to_path(uri)[source]

Convert a file:// URI back to a filesystem path.

Return type:

str

oats.lsp.client.language_id_for_path(path)[source]

Map a file extension to its LSP language ID.

Return type:

str

oats.lsp.client.detect_server_command(path)[source]

Detect the LSP server command for a file based on its extension.

Return type:

list[str] | None

oats.lsp.client.detect_workspace_server_command(root_dir)[source]

Detect the best LSP server command for a workspace root.

Return type:

list[str] | None

class oats.lsp.client.LSPServerInstance(root_dir, command, process=None, stdout=None, stdin=None, next_id=1, initialized=False, file_versions=<factory>, diagnostics_by_uri=<factory>)[source]

Bases: object

Represents a running LSP server process and its state.

root_dir: Path
command: list[str]
process: Process | None = None
stdout: StreamReader | None = None
stdin: StreamWriter | None = None
next_id: int = 1
initialized: bool = False
file_versions: dict[str, int]
diagnostics_by_uri: dict[str, list[dict[str, Any]]]
async start()[source]

Start the LSP server subprocess and initialize it.

Return type:

None

async request(method, params)[source]

Send a JSON-RPC request and wait for the matching response.

Return type:

Any

async notify(method, params)[source]

Send a JSON-RPC notification (no response expected).

Return type:

None

async sync_file(path, content)[source]

Sync file contents to the LSP server (didOpen or didChange).

Return type:

None

async collect_diagnostics(path, wait_timeout_s=0.75)[source]

Best-effort collection of diagnostics after a sync/open/change.

Many servers publish diagnostics asynchronously, so this polls cached notifications briefly instead of assuming they arrive immediately.

Return type:

list[dict[str, Any]]

__init__(root_dir, command, process=None, stdout=None, stdin=None, next_id=1, initialized=False, file_versions=<factory>, diagnostics_by_uri=<factory>)
class oats.lsp.client.LSPManager[source]

Bases: object

Small manager that reuses server processes per root + command.

__init__()[source]
get_instance(root_dir, command)[source]

Get or create an LSP server instance for the given root and command.

Return type:

LSPServerInstance

oats.lsp.client.get_lsp_manager()[source]

Return the process-wide LSP manager singleton.

Return type:

LSPManager

async oats.lsp.client.sync_file_if_supported(root_dir, path, content)[source]

Best-effort LSP sync for a file after edits/writes.

Returns True when a compatible language server exists and was updated.

Return type:

bool

oats.lsp.client.format_diagnostics_summary(path, diagnostics, limit=5)[source]

Format cached LSP diagnostics into a compact tool-friendly summary.

Return type:

str | None

async oats.lsp.client.sync_file_and_collect_diagnostics(root_dir, path, content)[source]

Best-effort helper used by write/edit paths.

Returns (lsp_synced, diagnostics_summary).

Return type:

tuple[bool, str | None]