From 70de7821bbace3d9234b962e29576f7c9be01a1d Mon Sep 17 00:00:00 2001
From: jahwag <540380+jahwag@users.noreply.github.com>
Date: Sun, 16 Nov 2025 08:49:00 +0100
Subject: [PATCH] feat: add claude code web sessions API

---
 pyproject.toml                             |   2 +-
 src/claudesync/cli/main.py                 |   2 +
 src/claudesync/cli/session.py              | 626 +++++++++++++++++++++
 src/claudesync/providers/base_claude_ai.py | 215 +++++++
 src/claudesync/providers/claude_ai.py      |  58 +-
 tests/mock_http_server.py                  |  67 ++-
 tests/test_claude_ai.py                    |  89 +++
 7 files changed, 1055 insertions(+), 4 deletions(-)
 create mode 100644 src/claudesync/cli/session.py

diff --git a/pyproject.toml b/pyproject.toml
index 52716c422..8d4f81b3e 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -1,6 +1,6 @@
 [project]
 name = "claudesync"
-version = "0.7.3"
+version = "0.7.4"
 authors = [
     {name = "Jahziah Wagner", email = "540380+jahwag@users.noreply.github.com"},
 ]
diff --git a/src/claudesync/cli/main.py b/src/claudesync/cli/main.py
index cc42656b2..8219ee66f 100644
--- a/src/claudesync/cli/main.py
+++ b/src/claudesync/cli/main.py
@@ -21,6 +21,7 @@
 from .project import project
 from .sync import schedule
 from .config import config
+from .session import session
 import logging
 
 logging.basicConfig(
@@ -252,6 +253,7 @@ def embedding(config, category, uberproject):
 cli.add_command(schedule)
 cli.add_command(config)
 cli.add_command(chat)
+cli.add_command(session)
 
 if __name__ == "__main__":
     cli()
diff --git a/src/claudesync/cli/session.py b/src/claudesync/cli/session.py
new file mode 100644
index 000000000..7594d4dff
--- /dev/null
+++ b/src/claudesync/cli/session.py
@@ -0,0 +1,626 @@
+import click
+from datetime import datetime
+from ..utils import handle_errors, validate_and_get_provider
+from ..exceptions import ProviderError
+
+
+@click.group()
+def session():
+    """Manage Claude Code web sessions."""
+    pass
+
+
+@session.group()
+def environment():
+    """Manage Claude Code environments."""
+    pass
+
+
+@session.group()
+def branch():
+    """Manage Claude Code repository branches."""
+    pass
+
+
+@branch.command(name="ls")
+@click.option(
+    "--json",
+    "json_output",
+    is_flag=True,
+    help="Output in JSON format",
+)
+@click.option(
+    "-s",
+    "--search",
+    help="Filter repositories by name",
+)
+@click.pass_obj
+@handle_errors
+def branch_ls(config, json_output, search):
+    """List available repositories for Claude Code sessions."""
+    import json as json_module
+
+    provider = validate_and_get_provider(config)
+    active_organization_id = config.get("active_organization_id")
+
+    try:
+        repos_data = provider.get_code_repos(active_organization_id)
+
+        if not repos_data or not repos_data.get("repos"):
+            click.echo("No repositories found.")
+            return
+
+        repos = repos_data["repos"]
+
+        # Filter by search term if provided
+        if search:
+            repos = [
+                r
+                for r in repos
+                if search.lower() in r.get("repo", {}).get("name", "").lower()
+            ]
+
+        if not repos:
+            click.echo(f"No repositories found matching '{search}'.")
+            return
+
+        if json_output:
+            click.echo(json_module.dumps(repos, indent=2))
+            return
+
+        # Display repositories in a formatted way
+        click.echo(f"Found {len(repos)} repository(ies):")
+
+        for idx, repo_data in enumerate(repos, 1):
+            repo = repo_data.get("repo", {})
+            name = repo.get("name", "Unknown")
+            owner = repo.get("owner", {}).get("login", "Unknown")
+            default_branch = repo.get("default_branch", "N/A")
+
+            click.echo(f"\n{idx}. {owner}/{name}")
+            click.echo(f"  Default branch: {default_branch}")
+
+    except ProviderError as e:
+        click.echo(f"Failed to list repositories: {str(e)}")
+
+
+@environment.command(name="ls")
+@click.option(
+    "--json",
+    "json_output",
+    is_flag=True,
+    help="Output in JSON format",
+)
+@click.pass_obj
+@handle_errors
+def environment_ls(config, json_output):
+    """List all Claude Code environments."""
+    import json as json_module
+
+    provider = validate_and_get_provider(config)
+    active_organization_id = config.get("active_organization_id")
+
+    try:
+        environments_data = provider.get_environments(active_organization_id)
+
+        if not environments_data or not environments_data.get("environments"):
+            click.echo("No environments found.")
+            return
+
+        environments = environments_data["environments"]
+
+        if json_output:
+            click.echo(json_module.dumps(environments, indent=2))
+            return
+
+        # Display environments in a formatted way
+        click.echo(f"Found {len(environments)} environment(s):")
+
+        for idx, env in enumerate(environments, 1):
+            env_id = env.get("environment_id", "N/A")
+            name = env.get("name", "Unnamed Environment")
+            kind = env.get("kind", "N/A")
+            state = env.get("state", "unknown")
+
+            click.echo(f"\n{idx}. {name}")
+            click.echo(f"  ID: {env_id}")
+            click.echo(f"  Kind: {kind}")
+            click.echo(f"  State: {state}")
+
+    except ProviderError as e:
+        click.echo(f"Failed to list environments: {str(e)}")
+
+
+@session.command()
+@click.option(
+    "-a",
+    "--all",
+    "archive_all",
+    is_flag=True,
+    help="Archive all active sessions",
+)
+@click.option(
+    "-y",
+    "--yes",
+    is_flag=True,
+    help="Skip confirmation prompt",
+)
+@click.pass_obj
+@handle_errors
+def archive(config, archive_all, yes):
+    """Archive existing sessions."""
+    provider = validate_and_get_provider(config)
+    active_organization_id = config.get("active_organization_id")
+    sessions_data = provider.get_sessions(active_organization_id)
+
+    if not sessions_data or not sessions_data.get("data"):
+        click.echo("No sessions found.")
+        return
+
+    # Filter for active sessions (not archived)
+    sessions = [
+        s
+        for s in sessions_data["data"]
+        if s.get("session_status") in ["running", "idle"]
+    ]
+
+    if not sessions:
+        click.echo("No active sessions found.")
+        return
+
+    if archive_all:
+        if not yes:
+            click.echo("The following sessions will be archived:")
+            for sess in sessions:
+                title = sess.get("title", "Untitled")
+                sess_id = sess.get("id", "N/A")
+                click.echo(f"  - {title} (ID: {sess_id})")
+            if not click.confirm("Are you sure you want to archive all sessions?"):
+                click.echo("Operation cancelled.")
+                return
+
+        success_count = 0
+        failure_count = 0
+        with click.progressbar(
+            sessions,
+            label="Archiving sessions",
+            item_show_func=lambda s: s.get("title", "Untitled") if s else "",
+        ) as bar:
+            for sess in bar:
+                try:
+                    provider.archive_session(active_organization_id, sess.get("id"))
+                    success_count += 1
+                except ProviderError as e:
+                    failure_count += 1
+                    title = sess.get("title", "Untitled")
+                    click.echo(f"\nFailed to archive session '{title}': {str(e)}")
+
+        click.echo(
+            f"\nArchive operation completed. "
+            f"Successfully archived: {success_count}, Failed: {failure_count}"
+        )
+        return
+
+    single_session_archival(sessions, yes, provider, active_organization_id)
+
+
+def single_session_archival(sessions, yes, provider, organization_id):
+    """Archive a single session selected by the user."""
+    click.echo("Available sessions to archive:")
+    for idx, sess in enumerate(sessions, 1):
+        title = sess.get("title", "Untitled")
+        sess_id = sess.get("id", "N/A")
+        click.echo(f"  {idx}. {title} (ID: {sess_id})")
+
+    selection = click.prompt("Enter the number of the session to archive", type=int)
+    if 1 <= selection <= len(sessions):
+        selected_session = sessions[selection - 1]
+        title = selected_session.get("title", "Untitled")
+        if yes or click.confirm(
+            f"Are you sure you want to archive the session '{title}'? "
+            f"Archived sessions cannot be modified but can still be viewed."
+        ):
+            try:
+                provider.archive_session(organization_id, selected_session.get("id"))
+                click.echo(f"Session '{title}' has been archived.")
+            except ProviderError as e:
+                click.echo(f"Failed to archive session '{title}': {str(e)}")
+    else:
+        click.echo("Invalid selection. Please try again.")
+
+
+@session.command()
+@click.argument("title", required=False)
+@click.option(
+    "-e",
+    "--environment-id",
+    help="Environment ID (if not provided, will try to use active environment)",
+)
+@click.option(
+    "-m",
+    "--model",
+    default="claude-sonnet-4-5-20250929",
+    help="Model to use (default: claude-sonnet-4-5-20250929)",
+)
+@click.option(
+    "-b",
+    "--branch",
+    help="Branch name to create (auto-generated if not provided)",
+)
+@click.option(
+    "--json",
+    "json_output",
+    is_flag=True,
+    help="Output in JSON format",
+)
+@click.pass_obj
+@handle_errors
+def create(config, title, environment_id, model, branch, json_output):  # noqa: C901
+    """Create a new Claude Code web session.
+
+    Provide a title for the session. If no title is provided, you will be prompted.
+    If the current directory is a git repository, it will be automatically linked to the session.
+    """
+    import json as json_module
+    import subprocess
+    import re
+
+    provider = validate_and_get_provider(config)
+    active_organization_id = config.get("active_organization_id")
+
+    # Get the title from user if not provided
+    if not title:
+        title = click.prompt("Enter the session title")
+
+    if not title.strip():
+        click.echo("Error: Title cannot be empty.")
+        return
+
+    # Get environment_id from config or parameter
+    if not environment_id:
+        environment_id = config.get("active_environment_id")
+        if not environment_id:
+            # Try to get environments and let user select
+            try:
+                environments_data = provider.get_environments(active_organization_id)
+                environments = environments_data.get("environments", [])
+
+                if not environments:
+                    click.echo("Error: No environments found.")
+                    click.echo(
+                        "Please create an environment first or use -e flag to specify one."
+                    )
+                    return
+
+                # Show available environments
+                click.echo("Available environments:")
+                for idx, env in enumerate(environments, 1):
+                    env_id = env.get("environment_id", "N/A")
+                    name = env.get("name", "Unnamed")
+                    state = env.get("state", "unknown")
+                    click.echo(f"  {idx}. {name} ({state}) - {env_id}")
+
+                # Prompt user to select
+                selection = click.prompt(
+                    "Select an environment number", type=int, default=1
+                )
+
+                if 1 <= selection <= len(environments):
+                    environment_id = environments[selection - 1].get("environment_id")
+                    if not json_output:
+                        click.echo(
+                            f"Using environment: {environments[selection - 1].get('name')}"
+                        )
+                else:
+                    click.echo("Invalid selection.")
+                    return
+
+            except ProviderError as e:
+                click.echo(f"Error: Could not retrieve environments: {str(e)}")
+                click.echo("Please use -e flag to specify an environment ID.")
+                return
+
+    # Try to detect git repository context and verify it's available
+    git_repo_url = None
+    git_repo_owner = None
+    git_repo_name = None
+    local_repo_detected = False
+    local_owner = None
+    local_name = None
+
+    try:
+        # Get git remote URL
+        result = subprocess.run(
+            ["git", "remote", "get-url", "origin"],
+            capture_output=True,
+            text=True,
+            check=True,
+        )
+        git_remote = result.stdout.strip()
+
+        # Parse GitHub URL (supports both SSH and HTTPS)
+        # SSH: git@github.com:owner/repo.git
+        # HTTPS: https://github.com/owner/repo.git
+        github_ssh_pattern = r"git@github\.com:([^/]+)/(.+?)(?:\.git)?$"
+        github_https_pattern = r"https://github\.com/([^/]+)/(.+?)(?:\.git)?$"
+
+        match = re.match(github_ssh_pattern, git_remote) or re.match(
+            github_https_pattern, git_remote
+        )
+        if match:
+            local_owner = match.group(1)
+            local_name = match.group(2)
+            local_repo_detected = True
+    except (subprocess.CalledProcessError, FileNotFoundError):
+        # Not in a git repository or git not available
+        local_repo_detected = False
+
+    # Get available repos to verify local repo is connected
+    repo_available = False
+    if local_repo_detected:
+        try:
+            repos_data = provider.get_code_repos(active_organization_id)
+            repos = repos_data.get("repos", [])
+
+            # Check if local repo is in available repos
+            for repo_data in repos:
+                repo = repo_data.get("repo", {})
+                owner = repo.get("owner", {}).get("login")
+                name = repo.get("name")
+                if owner == local_owner and name == local_name:
+                    git_repo_owner = local_owner
+                    git_repo_name = local_name
+                    git_repo_url = (
+                        f"https://github.com/{git_repo_owner}/{git_repo_name}"
+                    )
+                    repo_available = True
+                    if not json_output:
+                        click.echo(
+                            f"Using detected repository: {git_repo_owner}/{git_repo_name}"
+                        )
+                    break
+
+            if not repo_available and not json_output:
+                click.echo(
+                    f"\nDetected local repository {local_owner}/{local_name}, but it's not connected to Claude Code."
+                )
+                click.echo(
+                    "You need to connect this repository via GitHub OAuth first."
+                )
+                click.echo("Available repositories:")
+        except ProviderError:
+            repos = []
+
+    # If no valid repo, prompt user to select one
+    if not repo_available and not json_output:
+        try:
+            # If we haven't fetched repos yet, fetch them now
+            if not local_repo_detected or not repos:
+                repos_data = provider.get_code_repos(active_organization_id)
+                repos = repos_data.get("repos", [])
+
+            if repos:
+                for idx, repo_data in enumerate(repos, 1):
+                    repo = repo_data.get("repo", {})
+                    name = repo.get("name", "Unknown")
+                    owner = repo.get("owner", {}).get("login", "Unknown")
+                    click.echo(f"  {idx}. {owner}/{name}")
+                click.echo(
+                    f"  {len(repos) + 1}. Skip (create session without repository)"
+                )
+
+                selection = click.prompt(
+                    "Select a repository number", type=int, default=len(repos) + 1
+                )
+
+                if 1 <= selection <= len(repos):
+                    selected_repo = repos[selection - 1].get("repo", {})
+                    git_repo_owner = selected_repo.get("owner", {}).get("login")
+                    git_repo_name = selected_repo.get("name")
+                    if git_repo_owner and git_repo_name:
+                        git_repo_url = (
+                            f"https://github.com/{git_repo_owner}/{git_repo_name}"
+                        )
+                        click.echo(
+                            f"Using repository: {git_repo_owner}/{git_repo_name}"
+                        )
+                elif selection == len(repos) + 1:
+                    click.echo("Creating session without git repository context")
+                else:
+                    click.echo(
+                        "Invalid selection. Creating session without repository."
+                    )
+
+        except ProviderError:
+            # If we can't get repos, just continue without repo context
+            if not json_output:
+                click.echo("Creating session without git repository context")
+
+    try:
+        result = provider.create_session(
+            organization_id=active_organization_id,
+            title=title,
+            environment_id=environment_id,
+            git_repo_url=git_repo_url,
+            git_repo_owner=git_repo_owner,
+            git_repo_name=git_repo_name,
+            branch_name=branch,
+            model=model,
+        )
+
+        session_id = result.get("id", "N/A")
+        session_title = result.get("title", "N/A")
+        session_status = result.get("session_status", "N/A")
+
+        # Extract branch name from outcomes
+        branch_info = "N/A"
+        try:
+            outcomes = result.get("session_context", {}).get("outcomes", [])
+            for outcome in outcomes:
+                if outcome.get("type") == "git_repository":
+                    branches = outcome.get("git_info", {}).get("branches", [])
+                    if branches:
+                        branch_info = branches[0]
+        except (AttributeError, TypeError, KeyError):
+            pass
+
+        if json_output:
+            click.echo(json_module.dumps(result, indent=2))
+        else:
+            click.echo("Session created successfully!")
+            click.echo(f"ID: {session_id}")
+            click.echo(f"Title: {session_title}")
+            click.echo(f"Status: {session_status}")
+            if branch_info != "N/A":
+                click.echo(f"Branch: {branch_info}")
+
+            click.echo(f"\nView session at: https://claude.ai/code/{session_id}")
+            click.echo(
+                "\nNote: Session starts idle. Send a message through the web UI to begin."
+            )
+            click.echo("\n--- Streaming session events (Ctrl+C to stop) ---\n")
+            click.echo("Connecting to event stream...")
+
+            # Stream session events
+            try:
+                event_count = 0
+                for event in provider.stream_session_events(
+                    active_organization_id, session_id
+                ):
+                    event_count += 1
+
+                    # Debug: show raw events for now
+                    click.echo(f"[Event {event_count}] {json_module.dumps(event)}")
+
+                    if "error" in event:
+                        click.echo(f"Error: {event['error']}")
+                        break
+
+                    # Handle different event types
+                    event_type = event.get("type")
+                    if event_type == "message":
+                        content = event.get("content", "")
+                        if content:
+                            click.echo(f"Claude: {content}")
+                    elif event_type == "session_status":
+                        status = event.get("status", "")
+                        if status:
+                            click.echo(f"Status: {status}")
+
+                if event_count == 0:
+                    click.echo("\nNo events received from session.")
+                    click.echo("The session may still be initializing.")
+            except KeyboardInterrupt:
+                click.echo("\n\nSession streaming stopped by user.")
+                click.echo(f"Session {session_id} continues running in the background.")
+                click.echo(
+                    "You can view it at: https://claude.ai/code/session_{session_id}"
+                )
+            except Exception as e:
+                click.echo(f"\nError streaming events: {str(e)}")
+                click.echo(
+                    f"Session {session_id} is still running. View at: https://claude.ai/code/{session_id}"
+                )
+
+    except ProviderError as e:
+        click.echo(f"Failed to create session: {str(e)}")
+
+
+@session.command()
+@click.option(
+    "-a",
+    "--all",
+    "show_all",
+    is_flag=True,
+    help="Show all sessions including archived (default shows only running and idle)",
+)
+@click.option(
+    "--json",
+    "json_output",
+    is_flag=True,
+    help="Output in JSON format",
+)
+@click.pass_obj
+@handle_errors
+def ls(config, show_all, json_output):  # noqa: C901
+    """List all web sessions."""
+    import json as json_module
+
+    provider = validate_and_get_provider(config)
+    active_organization_id = config.get("active_organization_id")
+    sessions_data = provider.get_sessions(active_organization_id)
+
+    if not sessions_data or not sessions_data.get("data"):
+        click.echo("No sessions found.")
+        return
+
+    sessions = sessions_data["data"]
+
+    # Filter sessions if not showing all
+    if not show_all:
+        sessions = [
+            s for s in sessions if s.get("session_status") in ["running", "idle"]
+        ]
+
+    if not sessions:
+        click.echo("No active sessions found. Use --all to show archived sessions.")
+        return
+
+    if json_output:
+        click.echo(json_module.dumps(sessions, indent=2))
+        return
+
+    # Display sessions in a formatted way
+    click.echo(f"Found {len(sessions)} session(s):")
+
+    for idx, sess in enumerate(sessions, 1):
+        session_id = sess.get("id", "N/A")
+        title = sess.get("title", "Untitled")
+        status = sess.get("session_status", "unknown")
+        created_at = sess.get("created_at", "N/A")
+        updated_at = sess.get("updated_at", "N/A")
+
+        # Parse and format timestamps
+        try:
+            created_dt = datetime.fromisoformat(created_at.replace("Z", "+00:00"))
+            created_str = created_dt.strftime("%Y-%m-%d %H:%M:%S")
+        except (ValueError, AttributeError):
+            created_str = created_at
+
+        try:
+            updated_dt = datetime.fromisoformat(updated_at.replace("Z", "+00:00"))
+            updated_str = updated_dt.strftime("%Y-%m-%d %H:%M:%S")
+        except (ValueError, AttributeError):
+            updated_str = updated_at
+
+        # Get repository info if available
+        repo_info = ""
+        try:
+            context = sess.get("session_context", {})
+            outcomes = context.get("outcomes", [])
+            for outcome in outcomes:
+                if outcome.get("type") == "git_repository":
+                    git_info = outcome.get("git_info", {})
+                    repo = git_info.get("repo", "")
+                    branches = git_info.get("branches", [])
+                    if repo:
+                        repo_info = f"\n  Repository: {repo}"
+                        if branches:
+                            repo_info += f"\n  Branch: {branches[0]}"
+        except (AttributeError, TypeError, KeyError):
+            # Skip repo info if structure is unexpected
+            pass
+
+        # Status text
+        status_text = status.capitalize()
+
+        click.echo(f"\n{idx}. {title}")
+        click.echo(f"  ID: {session_id}")
+        click.echo(f"  Status: {status_text}")
+        click.echo(f"  Created: {created_str}")
+        click.echo(f"  Updated: {updated_str}")
+        if repo_info:
+            click.echo(repo_info)
+
+
+__all__ = ["session"]
diff --git a/src/claudesync/providers/base_claude_ai.py b/src/claudesync/providers/base_claude_ai.py
index 4da423b57..b3a66ca0a 100644
--- a/src/claudesync/providers/base_claude_ai.py
+++ b/src/claudesync/providers/base_claude_ai.py
@@ -265,6 +265,123 @@ def delete_chat(self, organization_id, conversation_uuids):
         data = {"conversation_uuids": conversation_uuids}
         return self._make_request("POST", endpoint, data)
 
+    def get_sessions(self, organization_id):
+        """Get all web sessions from the v1 API endpoint."""
+        # The sessions endpoint is at /v1/sessions, not under /api
+        # Requires x-organization-uuid header
+        return self._make_request_v1(
+            "GET", "/v1/sessions", organization_id=organization_id
+        )
+
+    def get_environments(self, organization_id):
+        """Get all environments from the v1 API endpoint."""
+        # Get environments for the organization
+        endpoint = f"/v1/environment_providers/private/organizations/{organization_id}/environments"
+        return self._make_request_v1("GET", endpoint, organization_id=organization_id)
+
+    def get_code_repos(self, organization_id, skip_status=True):
+        """Get all code repositories available for Claude Code sessions.
+
+        Args:
+            organization_id: The organization UUID
+            skip_status: Whether to skip fetching repository status (default: True)
+
+        Returns:
+            dict: Contains 'repos' array with repository information
+        """
+        endpoint = f"/organizations/{organization_id}/code/repos"
+        params = "?skip_status=true" if skip_status else ""
+        return self._make_request("GET", f"{endpoint}{params}")
+
+    def archive_session(self, organization_id, session_id):
+        """Archive a session by its ID."""
+        # Requires x-organization-uuid header
+        return self._make_request_v1(
+            "POST",
+            f"/v1/sessions/{session_id}/archive",
+            organization_id=organization_id,
+        )
+
+    def create_session(
+        self,
+        organization_id,
+        title,
+        environment_id,
+        git_repo_url=None,
+        git_repo_owner=None,
+        git_repo_name=None,
+        branch_name=None,
+        model="claude-sonnet-4-5-20250929",
+    ):
+        """Create a new Claude Code web session.
+
+        Args:
+            organization_id: The organization UUID
+            title: Session title
+            environment_id: Environment UUID (e.g., env_011CUPDTyMiRVMf18tfu2VUa)
+            git_repo_url: Optional git repository URL
+            git_repo_owner: Optional git repository owner (e.g., "Bytelope")
+            git_repo_name: Optional git repository name (e.g., "uppdragsradarn3")
+            branch_name: Optional branch name to create
+            model: Model to use (default: claude-sonnet-4-5-20250929)
+
+        Returns:
+            dict: Created session with id, title, session_context, etc.
+        """
+        endpoint = "/v1/sessions"
+
+        data = {
+            "title": title,
+            "environment_id": environment_id,
+            "session_context": {"model": model},
+        }
+
+        # Add git repository source if URL provided
+        if git_repo_url:
+            data["session_context"]["sources"] = [
+                {"type": "git_repository", "url": git_repo_url}
+            ]
+
+        # Add git repository outcome if repo details provided
+        if git_repo_owner and git_repo_name:
+            # If no branch name specified, generate a simple one
+            # The API will append the session ID automatically
+            if not branch_name:
+                # Generate from title: lowercase, replace spaces/special chars with hyphens
+                import re
+
+                safe_title = re.sub(r"[^a-z0-9]+", "-", title.lower()).strip("-")
+                # Limit to reasonable length
+                safe_title = safe_title[:50]
+                branch_name = f"claude/{safe_title}"
+
+            git_info = {
+                "type": "github",
+                "repo": f"{git_repo_owner}/{git_repo_name}",
+                "branches": [branch_name],
+            }
+
+            data["session_context"]["outcomes"] = [
+                {"type": "git_repository", "git_info": git_info}
+            ]
+
+        return self._make_request_v1(
+            "POST", endpoint, data=data, organization_id=organization_id
+        )
+
+    def _make_request_v1(self, method, endpoint, data=None, organization_id=None):
+        """Make a request to the v1 API (not under /api prefix).
+
+        Args:
+            method: HTTP method (GET, POST, etc.)
+            endpoint: API endpoint path
+            data: Optional request data
+            organization_id: Optional organization UUID for x-organization-uuid header
+        """
+        # This method should be implemented by subclasses to handle v1 API requests
+        # that are not under the /api prefix
+        raise NotImplementedError("This method should be implemented by subclasses")
+
     def _make_request(self, method, endpoint, data=None):
         raise NotImplementedError("This method should be implemented by subclasses")
 
@@ -292,6 +409,11 @@ def _make_request_stream(self, method, endpoint, data=None):
         # that can be used with sseclient
         raise NotImplementedError("This method should be implemented by subclasses")
 
+    def _make_request_stream_v1(self, method, endpoint, organization_id=None):
+        """Make a streaming request to the v1 API."""
+        # This method should be implemented by subclasses
+        raise NotImplementedError("This method should be implemented by subclasses")
+
     def send_message(
         self, organization_id, chat_id, prompt, timezone="UTC", model=None
     ):
@@ -319,3 +441,96 @@ def send_message(
                 yield {"error": event.data}
             if event.event == "done":
                 break
+
+    def _parse_sse_event(self, event):
+        """Parse a single SSE event and return the data."""
+        if not event.data or event.data.strip() == "":
+            return None
+        try:
+            return json.loads(event.data)
+        except json.JSONDecodeError:
+            self.logger.warning(f"Failed to parse event data: {event.data}")
+            return {"error": "Failed to parse JSON", "raw_data": event.data}
+
+    def stream_session_events(self, organization_id, session_id):
+        """Stream events from a Claude Code session.
+
+        Args:
+            organization_id: The organization UUID
+            session_id: The session ID to stream events from
+
+        Yields:
+            dict: Event data from the session stream
+        """
+        import signal
+
+        endpoint = f"/v1/sessions/{session_id}/events"
+        self.logger.debug(f"Opening SSE stream to {endpoint}")
+
+        def timeout_handler(signum, frame):
+            raise TimeoutError("No events received within timeout period")
+
+        response = self._make_request_stream_v1("GET", endpoint, organization_id)
+        client = sseclient.SSEClient(response)
+
+        # Set timeout for first event
+        old_handler = signal.signal(signal.SIGALRM, timeout_handler)
+        signal.alarm(30)
+
+        try:
+            for event_num, event in enumerate(client.events()):
+                if event_num == 0:
+                    signal.alarm(0)
+                    signal.signal(signal.SIGALRM, old_handler)
+
+                parsed_data = self._parse_sse_event(event)
+                if parsed_data:
+                    yield parsed_data
+
+                if event.event in ("error", "done"):
+                    if event.event == "error":
+                        yield {"error": event.data}
+                    break
+        except TimeoutError:
+            yield {
+                "error": "timeout",
+                "message": "No events received from session within 30 seconds",
+            }
+        finally:
+            signal.alarm(0)
+            signal.signal(signal.SIGALRM, old_handler)
+
+    def send_session_input(self, organization_id, session_id, prompt):
+        """Send input/prompt to a Claude Code session.
+
+        This is used to send an initial prompt or user input to a session.
+        The session will process the input and emit events through the event stream.
+
+        Args:
+            organization_id: The organization UUID
+            session_id: The session ID
+            prompt: The text prompt to send to Claude
+
+        Returns:
+            dict: Response from the API (typically session state or acknowledgment)
+        """
+        # Try different possible endpoints - the actual endpoint is not documented
+        possible_endpoints = [
+            (f"/v1/sessions/{session_id}/prompt", {"prompt": prompt}),
+            (f"/v1/sessions/{session_id}/message", {"message": prompt}),
+            (f"/v1/sessions/{session_id}/messages", {"content": prompt}),
+            (f"/v1/sessions/{session_id}/input", {"input": prompt}),
+        ]
+
+        last_error = None
+        for endpoint, data in possible_endpoints:
+            try:
+                return self._make_request_v1("POST", endpoint, data, organization_id)
+            except Exception as e:
+                last_error = e
+                # Try next endpoint
+                continue
+
+        # If all endpoints failed, raise the last error
+        if last_error:
+            raise last_error
diff --git a/src/claudesync/providers/claude_ai.py b/src/claudesync/providers/claude_ai.py
index 80587c2df..b3ff03e20 100644
--- a/src/claudesync/providers/claude_ai.py
+++ b/src/claudesync/providers/claude_ai.py
@@ -12,14 +12,20 @@ class ClaudeAIProvider(BaseClaudeAIProvider):
     def __init__(self, config=None):
         super().__init__(config)
 
-    def _make_request(self, method, endpoint, data=None):
-        url = f"{self.base_url}{endpoint}"
+    def _make_request_internal(  # noqa: C901
+        self, method, endpoint, data, base_url, extra_headers=None
+    ):
+        """Internal method to make HTTP requests with specified base URL."""
+        url = f"{base_url}{endpoint}"
         headers = {
             "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:129.0) Gecko/20100101 Firefox/129.0",
             "Content-Type": "application/json",
             "Accept-Encoding": "gzip",
         }
 
+        if extra_headers:
+            headers.update(extra_headers)
+
         session_key, expiry = self.config.get_session_key("claude.ai")
         cookies = {
             "sessionKey": session_key,
@@ -75,6 +81,9 @@ def _make_request(self, method, endpoint, data=None):
             self.logger.error(f"Response content: {content_str}")
             raise ProviderError(f"Invalid JSON response from API: {str(json_err)}")
 
+    def _make_request(self, method, endpoint, data=None):
+        return self._make_request_internal(method, endpoint, data, self.base_url)
+
     def handle_http_error(self, e):
         self.logger.debug(f"Request failed: {str(e)}")
         self.logger.debug(f"Response status code: {e.code}")
@@ -116,6 +125,24 @@ def handle_http_error(self, e):
             self.logger.error(error_msg)
             raise ProviderError(error_msg)
 
+    def _make_request_v1(self, method, endpoint, data=None, organization_id=None):
+        """Make a request to the v1 API (not under /api prefix)."""
+        # For v1 endpoints, we use the root URL without the /api prefix
+        base_url = self.base_url.replace("/api", "")
+
+        # Add required Anthropic headers for v1 API
+        extra_headers = {
+            "anthropic-version": "2023-06-01",
+        }
+
+        # Add organization header if provided
+        if organization_id:
+            extra_headers["x-organization-uuid"] = organization_id
+
+        return self._make_request_internal(
+            method, endpoint, data, base_url, extra_headers
+        )
+
     def _make_request_stream(self, method, endpoint, data=None):
         url = f"{self.base_url}{endpoint}"
         session_key, _ = self.config.get_session_key("claude.ai")
@@ -136,3 +163,30 @@ def _make_request_stream(self, method, endpoint, data=None):
             self.handle_http_error(e)
         except urllib.error.URLError as e:
             raise ProviderError(f"API request failed: {str(e)}")
+
+    def _make_request_stream_v1(self, method, endpoint, organization_id=None):
+        """Make a streaming request to the v1 API."""
+        # For v1 endpoints, use root URL without /api prefix
+        base_url = self.base_url.replace("/api", "")
+        url = f"{base_url}{endpoint}"
+
+        session_key, _ = self.config.get_session_key("claude.ai")
+        headers = {
+            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:129.0) Gecko/20100101 Firefox/129.0",
+            "Accept": "text/event-stream",
+            "anthropic-version": "2023-06-01",
+            "Cookie": f"sessionKey={session_key}",
+        }
+
+        # Add organization header if provided
+        if organization_id:
+            headers["x-organization-uuid"] = organization_id
+
+        req = urllib.request.Request(url, method=method, headers=headers)
+
+        try:
+            return urllib.request.urlopen(req)
+        except urllib.error.HTTPError as e:
+            self.handle_http_error(e)
+        except urllib.error.URLError as e:
+            raise ProviderError(f"API request failed: {str(e)}")
diff --git a/tests/mock_http_server.py b/tests/mock_http_server.py
index 8f1ad5a57..dfbfdb55e 100644
--- a/tests/mock_http_server.py
+++ b/tests/mock_http_server.py
@@ -34,6 +34,21 @@ def do_GET(self):
                 }
             )
             self.wfile.write(response.encode())
+        elif "/events" in parsed_path.path and "/sessions/" in parsed_path.path:
+            # Handle session events streaming
+            self.send_response(200)
+            self.send_header("Content-type", "text/event-stream")
+            self.send_header("Cache-Control", "no-cache")
+            self.end_headers()
+            # Send mock session events
+            self.wfile.write(b'data: {"type":"session_status","status":"running"}\n\n')
+            self.wfile.write(
+                b'data: {"type":"message","content":"Starting Claude Code session..."}\n\n'
+            )
+            self.wfile.write(
+                b'data: {"type":"message","content":"Environment initialized"}\n\n'
+            )
+            self.wfile.write(b"event: done\n\n")
         else:
             print(f"Received GET request: {self.path}")
             # time.sleep(0.01)  # Add a small delay to simulate network latency
@@ -94,7 +109,7 @@ def do_GET(self):
             else:
                 self.send_error(404, "Not Found")
 
-    def do_POST(self):
+    def do_POST(self):  # noqa: C901
         content_length = int(self.headers["Content-Length"])
         parsed_path = urlparse(self.path)
 
@@ -114,6 +129,56 @@ def do_POST(self):
                 b'data: {"completion": "I apologize for the confusion. You\'re right."}\n\n'
             )
             self.wfile.write(b"event: done\n\n")
+        elif parsed_path.path == "/v1/sessions":
+            # Handle session creation
+            post_data = self.rfile.read(content_length)
+            data = json.loads(post_data.decode("utf-8"))
+
+            title = data.get("title", "Test Session")
+            environment_id = data.get("environment_id", "env_test")
+            session_context = data.get("session_context", {})
+
+            # Create mock session response
+            session_id = "session_test123"
+            response_data = {
+                "id": session_id,
+                "title": title,
+                "environment_id": environment_id,
+                "session_status": "running",
+                "session_context": session_context,
+                "type": "session",
+                "created_at": "2025-11-16T08:09:45.536149996Z",
+                "updated_at": "2025-11-16T08:09:45.536149996Z",
+            }
+
+            # If git repository outcomes were provided, ensure branch name is set
+            if "outcomes" in session_context:
+                for outcome in session_context["outcomes"]:
+                    if outcome.get("type") == "git_repository":
+                        git_info = outcome.get("git_info", {})
+                        if "branches" not in git_info or not git_info["branches"]:
+                            # Auto-generate branch name if not provided
+                            git_info["branches"] = [f"claude/test-session-{session_id}"]
+
+            self.send_response(200)
+            self.send_header("Content-type", "application/json")
+            self.end_headers()
+            self.wfile.write(json.dumps(response_data).encode())
+        elif "/sessions/" in parsed_path.path and parsed_path.path.endswith("/input"):
+            # Handle session input (sending prompt to session)
+            post_data = self.rfile.read(content_length)
+            data = json.loads(post_data.decode("utf-8"))
+
+            # Just acknowledge the input was received
+            response_data = {
+                "status": "accepted",
+                "input_received": data.get("input", ""),
+            }
+
+            self.send_response(200)
+            self.send_header("Content-type", "application/json")
+            self.end_headers()
+            self.wfile.write(json.dumps(response_data).encode())
         else:
             # time.sleep(0.01)  # Add a small delay to simulate network latency
             content_length = int(self.headers["Content-Length"])
diff --git a/tests/test_claude_ai.py b/tests/test_claude_ai.py
index 14dd2c847..40b39dc15 100644
--- a/tests/test_claude_ai.py
+++ b/tests/test_claude_ai.py
@@ -159,6 +159,95 @@ def test_handle_http_error_403(self):
             self.provider.handle_http_error(mock_error)
         self.assertIn("403 Forbidden error", str(context.exception))
 
+    def test_create_session_with_branch(self):
+        result = self.provider.create_session(
+            organization_id="org1",
+            title="Test Session",
+            environment_id="env_test",
+            git_repo_url="https://github.com/test/repo",
+            git_repo_owner="test",
+            git_repo_name="repo",
+            branch_name="claude/test-branch",
+        )
+        self.assertIn("id", result)
+        self.assertIn("title", result)
+        self.assertIn("session_status", result)
+        self.assertEqual(result["title"], "Test Session")
+        self.assertEqual(result["session_status"], "running")
+        self.assertEqual(result["environment_id"], "env_test")
+
+        # Check git repository context
+        outcomes = result.get("session_context", {}).get("outcomes", [])
+        self.assertTrue(len(outcomes) > 0)
+        git_outcome = outcomes[0]
+        self.assertEqual(git_outcome["type"], "git_repository")
+        self.assertIn("branches", git_outcome["git_info"])
+        self.assertEqual(git_outcome["git_info"]["branches"][0], "claude/test-branch")
+
+    def test_create_session_auto_branch(self):
+        # Test session creation with auto-generated branch name
+        result = self.provider.create_session(
+            organization_id="org1",
+            title="Auto Branch Session",
+            environment_id="env_test",
+            git_repo_url="https://github.com/test/repo",
+            git_repo_owner="test",
+            git_repo_name="repo",
+            # No branch_name specified - should auto-generate
+        )
+        self.assertEqual(result["title"], "Auto Branch Session")
+        self.assertEqual(result["session_status"], "running")
+
+        # Check that branch was auto-generated
+        outcomes = result.get("session_context", {}).get("outcomes", [])
+        self.assertTrue(len(outcomes) > 0)
+        git_outcome = outcomes[0]
+        self.assertEqual(git_outcome["type"], "git_repository")
+        self.assertIn("branches", git_outcome["git_info"])
+        # Branch should be auto-generated with session ID
+        self.assertTrue(git_outcome["git_info"]["branches"][0].startswith("claude/"))
+
+    def test_create_session_minimal(self):
+        # Test session creation with minimal parameters (no git context)
+        result = self.provider.create_session(
+            organization_id="org1",
+            title="Minimal Session",
+            environment_id="env_test",
+        )
+        self.assertEqual(result["title"], "Minimal Session")
+        self.assertEqual(result["session_status"], "running")
+        self.assertIn("session_context", result)
+        # Should not have outcomes or sources without git context
+        session_context = result.get("session_context", {})
+        self.assertNotIn("sources", session_context)
+        self.assertNotIn("outcomes", session_context)
+
+    def test_stream_session_events(self):
+        # Test streaming events from a session
+        events = list(
+            self.provider.stream_session_events(
+                organization_id="org1", session_id="session_test123"
+            )
+        )
+        # Should receive at least 3 events from mock server
+        self.assertGreaterEqual(len(events), 3)
+        # Check first event
+        self.assertEqual(events[0]["type"], "session_status")
+        self.assertEqual(events[0]["status"], "running")
+        # Check subsequent events
+        self.assertEqual(events[1]["type"], "message")
+        self.assertIn("Starting Claude Code", events[1]["content"])
+
+    def test_send_session_input(self):
+        # Test sending input/prompt to a session
+        result = self.provider.send_session_input(
+            organization_id="org1",
+            session_id="session_test123",
+            prompt="Hello, please help me fix a bug",
+        )
+        self.assertEqual(result["status"], "accepted")
+        self.assertEqual(result["input_received"], "Hello, please help me fix a bug")
+
 
 if __name__ == "__main__":
     unittest.main()