Core API

About file generation for Continuity projects.

This module provides backward compatibility import for AboutGenerator. The implementation has been moved to the generators submodule.

class continuity.core.about.AboutGenerator(project_path)

Generates ABOUT.md files from templates and changelog entries.

generate()

Generate ABOUT.md content from template and changelog.

Return type:

str

Returns:

Generated ABOUT.md content

write()

Generate and write ABOUT.md to continuity directory.

Return type:

Path

Returns:

Path to written ABOUT.md file

Constants for Continuity core functionality.

This module centralizes hardcoded values to make the system more maintainable.

Simple file processing for Continuity inbox. Phase 3: Simplified to handle only basic markdown file processing.

class continuity.core.file_processor.SimpleFileProcessor(base_path, config)

Simple file processor that handles markdown files only.

ensure_directories()

Create necessary directories.

Return type:

None

scan_inbox()

Scan inbox for markdown files to process.

Return type:

List[Path]

detect_recipient(content)

Detect recipient from content (simple @mention detection).

Return type:

str

extract_title(content, filename)

Extract title from content or use filename.

Return type:

str

process_file(file_path)

Process a single markdown file.

Return type:

Dict[str, Any]

archive_file(file_path)

Archive processed file.

Return type:

None

get_status()

Get simple inbox status.

Return type:

Dict[str, Any]

Inbox functionality for Continuity.

Provides automatic processing of files dropped into the inbox directory, adding metadata and moving them to the appropriate mailbox.

class continuity.core.inbox.InboxProcessor(base_path, config)

Processes files dropped into the inbox directory.

ensure_directories()

Create inbox directories if they don’t exist.

Return type:

None

scan_inbox()

Get list of files in inbox to process.

Return type:

List[Path]

Returns:

List of file paths to process

extract_metadata(content)

Extract frontmatter metadata and content from file.

Parameters:

content (str) – File content

Return type:

Tuple[Dict[str, Any], str]

Returns:

Tuple of (metadata dict, content without frontmatter)

infer_recipient(content, metadata)

Determine recipient from explicit metadata only.

Parameters:

• content (str) – Message content (unused, for compatibility)

• metadata (Dict[str, Any]) – Extracted metadata

Return type:

str

Returns:

Recipient username (defaults to ai_user for inbox processing)

extract_title(content, metadata, filename)

Extract or generate title for message.

Parameters:

• content (str) – Message content

• metadata (Dict[str, Any]) – Extracted metadata

• filename (str) – Original filename

Return type:

str

Returns:

Title string

process_file(filepath, archive=True)

Process a single file from inbox.

Parameters:

• filepath (Path) – Path to file to process

• archive (bool) – Whether to archive the original file

Return type:

Optional[Path]

Returns:

Path to created message file, or None if failed

process_all(archive=True)

Process all files in inbox.

Parameters:

archive (bool) – Whether to archive processed files

Return type:

Dict[str, Any]

Returns:

Dict with processing results

watch(callback=None, interval=2)

Watch inbox for new files (future enhancement).

Parameters:

• callback – Optional callback for each processed file

• interval (int) – Check interval in seconds

Return type:

None

Simple project initialization functionality for Continuity. Phase 3: Simplified to remove complex knowledge base and advanced features.

class continuity.core.init.ProjectInitializer(project_path, config)

Handles project initialization and setup.

detect_human_user()

Detect a sensible human username.

Return type:

str

Returns:

Username string

create_directory_structure()

Create the basic .continuity directory structure.

Return type:

Dict[str, Path]

Returns:

Dict mapping structure names to paths

create_cli_reference()

Create the CLI reference file.

Return type:

Path

Returns:

Path to created CLI reference

create_ai_agent_reference()

Create or update the AI Agent Reference file.

Return type:

Path

Returns:

Path to created/updated AI Agent Reference

create_project_config(human_user=None)

Create project-specific configuration.

Parameters:

human_user (Optional[str]) – Human username, auto-detected if None

Return type:

Path

Returns:

Path to created config file

create_about_file()

Create or update the ABOUT.md file.

Return type:

Path

Returns:

Path to created/updated ABOUT.md file

regenerate_all_documents()

Regenerate all project documents using the factory pattern.

Return type:

Dict[str, Path]

Returns:

Dict mapping document names to their file paths

create_gitignore_entry()

Add continuity directory to .gitignore if it exists.

Return type:

bool

Returns:

True if .gitignore was updated, False otherwise

create_welcome_message(human_user)

Create a welcome message for the human user.

Parameters:

human_user (str) – Human username

Return type:

Path

Returns:

Path to created message

init_project(force=False)

Initialize a new continuity project.

Parameters:

force (bool) – Overwrite existing .continuity directory

Return type:

Dict[str, Any]

Returns:

Dict with initialization results

doctor_project(auto_yes=False)

Non-destructive project doctor - ensure current structure exists without touching existing files.

Return type:

Dict[str, Any]

Returns:

Dict with doctor results

Core mailbox functionality for Continuity.

This module provides the main business logic for managing asynchronous messages between humans and AI assistants. It handles message storage, retrieval, and organization using various storage backends.

The module contains the core classes for message handling: - Message: Represents individual messages with metadata - MailboxManager: High-level mailbox operations

Example

Basic usage:

from continuity.core.mailbox import MailboxManager, Message
from continuity.config.base import ContinuityConfig

config = ContinuityConfig.load()
manager = MailboxManager(config)

# Create and send a message
message = Message(
    id="msg-001",
    content="Hello AI assistant!",
    timestamp=datetime.now(),
    sender="human_user",
    recipient="ai_user"
)

success = manager.send_message(message)

class continuity.core.mailbox.Message(id, content, timestamp, sender, recipient, read=False, title='', metadata=None)

Represents a message in the async mailbox system.

A Message encapsulates all the information needed for asynchronous communication between humans and AI assistants. Messages support rich metadata, read/unread status tracking, and serialization.

Parameters:

• id (str) – Unique identifier for the message

• content (str) – The main message content (supports Markdown)

• timestamp (datetime) – When the message was created

• sender (str) – Username of the message sender

• recipient (str) – Username of the message recipient

• read (bool) – Whether the message has been read (defaults to False)

• title (str) – Optional message title (extracted from content if not provided)

• metadata (Optional[Dict[str, Any]]) – Optional dictionary of additional metadata

id

Unique message identifier

Type:

str

content

Message content

Type:

str

timestamp

Creation timestamp

Type:

datetime

sender

Sender username

Type:

str

recipient

Recipient username

Type:

str

read

Read status

Type:

bool

title

Message title

Type:

str

metadata

Additional metadata

Type:

Dict[str, Any]

Example

Creating a simple message:

message = Message(
    id="msg-001",
    content="Hello Ada! How are you?",
    timestamp=datetime.now(),
    sender="human",
    recipient="agent"
)

Creating a message with metadata:

message = Message(
    id="msg-002",
    content="Please review the API design",
    timestamp=datetime.now(),
    sender="human_user",
    recipient="ai_user",
    title="API Review Request",
    metadata={
        "priority": "high",
        "project": "continuity",
        "tags": ["api", "review"]
    }
)

to_dict()

Serialize message to dictionary format.

Converts the message object to a dictionary suitable for JSON serialization or storage. The timestamp is converted to ISO format.

Return type:

Dict[str, Any]

Returns:

Dictionary containing all message fields with JSON-serializable values.

Example

>>> message = Message(id="test", content="Hello", ...)
>>> data = message.to_dict()
>>> print(data['timestamp'])
'2023-06-17T12:00:00'

classmethod from_file(filepath)

Load message from file with frontmatter parsing.

Reads a message from a markdown file with YAML frontmatter. The frontmatter can contain metadata like sender, recipient, title, etc. If no title is provided in frontmatter, it will be extracted from the first heading in the content or derived from the filename.

Parameters:

filepath (Path) – Path to the message file to load

Return type:

Message

Returns:

Message object parsed from the file

Raises:

• FileNotFoundError – If the specified file doesn’t exist

• PermissionError – If the file cannot be read

Example

File structure:

---
from: human
to: agent
priority: high
---

# API Review Request

Please review the new API design...

Loading the message:

>>> message = Message.from_file(Path("api-review.md"))
>>> print(message.sender)
'michael'
>>> print(message.title)
'API Review Request'

class continuity.core.mailbox.MailboxManager(base_path, config)

Core mailbox functionality.

check_mailbox(user=None)

Check mailbox for all messages (unread, read, and archived).

Return type:

List[Message]

get_unread_messages(user=None)

Get only unread messages.

Return type:

List[Message]

get_read_messages(user=None)

Get read messages from the read folder.

Return type:

List[Message]

get_message_by_id(user, message_id)

Get a specific message by ID from unread, read, and archive folders.

Return type:

Optional[Message]

send_message(content, to=None, from_user=None, title='')

Send a message to a mailbox.

Return type:

Message

mark_read(user, message_id)

Mark a message as read by moving it.

Return type:

bool

mark_unread(user, message_id)

Mark a message as unread by moving it back from read folder.

Return type:

bool

delete_message(user, message_id)

Delete a message permanently.

Return type:

bool

archive_message(user, message_id, from_read=False)

Archive a message by moving it to archive folder.

Return type:

bool

get_archived_messages(user)

Get all archived messages for a user.

Return type:

List[Message]

unarchive_message(user, message_id, to_unread=False)

Restore a message from archive.

Return type:

bool

class continuity.core.mailbox.ContinuityCore(base_path, config)

Core Continuity functionality.

say_hello(name='World')

Hello world with persistence.

Return type:

Dict[str, Any]

get_hello_history()

Get hello history.

Return type:

List[Dict[str, Any]]

get_status(max_messages=5, timeout_seconds=10)

Get overall continuity status with safety limits.

Parameters:

• max_messages (int) – Maximum recent messages to include

• timeout_seconds (int) – Maximum time to spend gathering status

Return type:

Dict[str, Any]

prime_context(project_path=None)

Prime context with project information and communication updates.

Return type:

Dict[str, Any]

mark_message_read(message_id, user)

Mark a message as read.

Return type:

bool

Notification system for tracking new activity across threads and mailboxes.

This module implements the pull notification system for Continuity, providing gentle awareness of new activity without interruption.

class continuity.core.notifications.NotificationState(project_path, config=None)

Manages notification state for tracking new activity.

get_last_check()

Get last check timestamp.

Return type:

datetime

mark_checked()

Mark current time as last check.

Return type:

None

get_seen_posts(thread_name)

Get number of seen posts for a thread.

Return type:

int

mark_thread_seen(thread_name, post_count)

Mark a thread as seen up to a certain post count.

Return type:

None

get_notification_style()

Get user’s notification style preference.

Return type:

str

class continuity.core.notifications.ActivityDetector(notification_state)

Detects new activity in threads and mailboxes.

get_thread_activity(threads_path)

Get threads with new activity since last check.

Return type:

List[Dict[str, Any]]

get_mailbox_activity(mailbox_path, user)

Get unread message count for a user.

Return type:

int

Schema management for Continuity project structure.

class continuity.core.schema.ContinuitySchema(storage_path)

Manages .continuity directory structure and validation.

get_current_version()

Get the current schema version.

Return type:

Tuple[int, int, int, str, str]

validate_structure()

Validate the directory structure.

Return type:

Dict[str, Any]

create_structure()

Create the basic directory structure.

Return type:

List[str]

get_status()

Get schema status information.

Return type:

Dict[str, Any]