Utilities API

Emergency cleanup utilities for continuity plugin.

continuity.utils.cleanup.get_continuity_processes()

Get list of continuity process PIDs.

Return type:

List[int]

Returns:

List of process IDs running continuity

continuity.utils.cleanup.terminate_processes(pids, force=False)

Terminate continuity processes.

Parameters:

• pids (List[int]) – List of process IDs to terminate

• force (bool) – Use SIGKILL instead of SIGTERM

Return type:

bool

Returns:

True if all processes terminated successfully

continuity.utils.cleanup.clean_lock_files()

Clean up lock files.

Return type:

int

Returns:

Number of files removed

continuity.utils.cleanup.clean_socket_files()

Clean up socket files.

Return type:

int

Returns:

Number of files removed

continuity.utils.cleanup.clean_temp_files()

Clean up temporary files.

Return type:

int

Returns:

Number of files removed

continuity.utils.cleanup.emergency_cleanup(force=False)

Perform emergency cleanup of continuity resources.

Parameters:

force (bool) – Use aggressive cleanup (SIGKILL, etc.)

Return type:

bool

Returns:

True if cleanup was successful

Editor integration utilities for Continuity. Provides smart editor detection and positioning for frictionless workflow.

class continuity.utils.editor.EditorConfig

Configuration for different editor positioning support.

continuity.utils.editor.detect_editor()

Detect the best available editor.

Return type:

str

Returns:

Editor command string

continuity.utils.editor.get_editor_name(editor_command)

Extract the base editor name from a command.

Parameters:

editor_command (str) – Full editor command (may include flags)

Return type:

str

Returns:

Base editor name

continuity.utils.editor.build_editor_command(editor_command, file_path, line_number=1)

Build editor command with positioning.

Parameters:

• editor_command (str) – Editor command to use

• file_path (str) – Path to file to edit

• line_number (int) – Line number to position cursor

Return type:

List[str]

Returns:

Command list ready for subprocess

continuity.utils.editor.create_message_template(title='', to='agent', from_user='human', config=None)

Create a message template with frontmatter.

Parameters:

• title (str) – Message title

• to (str) – Recipient (internal username or alias)

• from_user (str) – Sender (internal username or alias)

• config – ContinuityConfig instance for display name mapping

Return type:

str

Returns:

Template content with positioning markers

continuity.utils.editor.find_content_line(template)

Find the line number where user should start typing.

Parameters:

template (str) – Template content

Return type:

int

Returns:

Line number (1-based) for cursor positioning

continuity.utils.editor.open_editor_with_template(template, editor_command=None)

Open editor with template and return content.

Parameters:

• template (str) – Template content to pre-fill

• editor_command (Optional[str]) – Specific editor to use (auto-detect if None)

Return type:

Optional[str]

Returns:

Edited content, or None if editing was cancelled

continuity.utils.editor.test_editor_positioning(editor_command=None)

Test editor positioning functionality.

Parameters:

editor_command (Optional[str]) – Editor to test (auto-detect if None)

Return type:

None

Pager utilities for enhanced content viewing. Provides smart pager detection and beautiful content formatting.

continuity.utils.pager.detect_pager()

Detect the best available pager.

Return type:

str

Returns:

Pager command string

continuity.utils.pager.format_message_content(message, show_metadata=True, user_context=None)

Format message content for display.

Parameters:

• message – Message object with content, metadata

• show_metadata (bool) – Whether to include metadata header

• user_context – UserContext for nickname mapping (optional)

Return type:

str

Returns:

Formatted content string

continuity.utils.pager.format_thread_content(thread_name, posts)

Format thread content for display.

Parameters:

• thread_name (str) – Name of the thread

• posts (list) – List of post objects

Return type:

str

Returns:

Formatted thread content

continuity.utils.pager.view_with_pager(content, title='Content')

Display content using system pager.

Parameters:

• content (str) – Content to display

• title (str) – Title for the content (used in temp filename)

Return type:

bool

Returns:

True if content was displayed successfully

continuity.utils.pager.preview_content(content, max_lines=10)

Create a preview of content for quick display.

Parameters:

• content (str) – Full content

• max_lines (int) – Maximum lines to show in preview

Return type:

str

Returns:

Preview content with ellipsis if truncated

Process lock system to prevent multiple server instances.

class continuity.utils.process_lock.ProcessLock(name='continuity_server')

Exclusive process lock to prevent multiple server instances.

acquire()

Acquire exclusive lock.

Return type:

bool

Returns:

True if lock acquired successfully, False if already locked

release()

Release lock and clean up.

Return type:

None

is_locked()

Check if lock is currently held.

Return type:

bool

Returns:

True if lock file exists and is locked

get_lock_pid()

Get PID of process holding the lock.

Return type:

Optional[int]

Returns:

PID of lock holder, or None if not locked or can’t read

__enter__()

Context manager entry.

__exit__(exc_type, exc_val, exc_tb)

Context manager exit.