Forem: Pieces 🌟

How to Build an Agentic Blog Generator

Pieces 🌟 — Tue, 03 Feb 2026 15:22:20 +0000

Building an Agentic Blog Generator With Pieces OS (Flutter)

This project is a Flutter app that generates a technical blog (in Markdown) from real, recent context. The core idea is simple: use Pieces OS as the source of truth for what you’ve been working on (workstream summaries + your own annotations/persona signals), then use an LLM to turn that into a structured, high-quality blog—optionally in agentic mode with MCP tools.

Features

Auto generate title: suggests blog titles from recent Pieces workstream summaries.

Detect persona: pulls recent Pieces user annotations and converts them into “persona signals” for voice/tone

Detect style: analyzes a sample blog to infer a reusable style object for consistent output.

Generate multi part blog: plans part titles + per-part outlines, then generates Markdown one part at a time.

Agentic flow: connects to the MCP endpoint and uses tools for retrieval/verification (RAG) instead of guessing.

What Pieces OS gives us (and why it matters)

When you ask an LLM to “write a blog about my project”, you usually get one of two outcomes:

A generic post that sounds plausible but doesn’t match what you actually did.
A post that misses the details that made the work interesting (trade-offs, decisions, workflow).

Pieces OS helps fix this by providing grounded context:

Workstream Summaries (LTM): a stream of recent summaries of your work so we can generate content from what actually happened recently.
Annotations (persona + preferences + voice cues): your own notes/annotations can be pulled and turned into “persona signals” so the blog tone and framing match you.
MCP (Model Context Protocol) tools: in agentic mode, the generator can call MCP tools (memory/search/context) instead of guessing—so it can fetch or verify information as it writes. Practically, connecting via the MCP endpoint gives us RAG (retrieval‑augmented generation): the model retrieves relevant context from Pieces and then writes with that grounded input.

In this app, Pieces is not a “nice-to-have integration”—it’s the backbone for relevance and personalization.

How this app uses Pieces OS

At a high level, the wizard does three Pieces-backed things:

Connect to Pieces OS
- Establishes app connection with Pieces OS (so API calls work).
Pull recent context
- Pulls recent workstream summary IDs over a short-lived WebSocket.
- Fetches each summary snapshot.
- Extracts the human-readable “DESCRIPTION” annotation text from each summary.
Pull persona signals
- Fetches recent user annotations via UserApi.userGetAnnotations().
- Normalizes/truncates them into a prompt-friendly “persona signals” block.

Then, when we run in agentic mode, we also:

Connect to the Pieces MCP endpoint
List the available tools
Allow the LLM to call those tools while planning/writing

This combination (summaries + persona + MCP tools) gives the generator a much better shot at producing content that matches reality.

Agentic mode: how we got better output (one-shot → plan-first → structured parts)

We saw a clear quality jump as we changed the prompting approach.

Attempt 1: one prompt to generate the whole blog

The naive approach is: “Here’s some context—write the entire blog.”

In practice, that tends to produce:

weak structure (rambling, repetitive sections)
missing coverage (important modules not mentioned)
brittle accuracy (model fills gaps by guessing)

Attempt 2: plan first, then generate

Next improvement: separate planning from writing.

Instead of writing immediately, we made the app generate a plan first (what parts, what each part is about). That makes the writing phase far more constrained and coherent:

the model knows the “shape” of the blog before it writes
you can review/edit titles before any heavy generation happens

Attempt 3: even more planning, broken into titled parts

The biggest jump came from adding more structure:

generate part titles first (so each part has a clear purpose)
generate an outline for each part (so headings and subheadings are defined)
then write one part at a time, using the outline as a contract

That does two things:

Focus: each part stays on-topic (because it has a title + outline).
Better coverage: parts are explicitly scoped, so important areas are less likely to be skipped.

This is exactly why titled parts matter: the model is no longer inventing structure as it writes—it’s executing a plan you’ve already approved.

Tutorial: integrating Pieces OS in a Flutter app

This section breaks down how we integrated Pieces OS in a way that works well for a UI-driven Flutter app. The goal is to make the model’s output grounded (Pieces workstream summaries + persona signals) and optionally agentic (MCP endpoint → RAG).

Step 1: configure the Pieces endpoints

In PiecesOSService, we keep the local Pieces defaults in one place:

baseUrl: REST API base (http://localhost:39300)
websocketUrl: WebSocket base (ws://localhost:39300)
defaultMcpEndpoint: MCP streamable HTTP endpoint (used for tool-based RAG)

Here’s the exact code that defines those endpoints (and the imports used by the service):

import 'dart:async';
import 'dart:developer' as dev;
import 'dart:convert';
import 'dart:io';
import 'package:mcp_dart/mcp_dart.dart' as mcp;
import 'package:pieces_os_client/api.dart';

import '../models/persona_signals.dart';

/// Service to interact with Pieces OS for LTM (Long Term Memory)
class PiecesOSService {
  // Pieces OS configuration
  static const String baseUrl = 'http://localhost:39300';
  static const String websocketUrl = 'ws://localhost:39300';
  static const String defaultMcpEndpoint =
      'http://localhost:39300/model_context_protocol/2025-03-26/mcp';

Step 2: connect the app to Pieces OS (REST)

The connectApplication() method registers/connects this app with Pieces via ConnectorApi.connect(...) and stores the returned Context.

The initialize() method is a small guard that:

ensures we only connect once per session (_isInitialized)
makes all later calls safe to run “just-in-time” from the UI

Here are the service fields + constructor + REST connection code:

  // API clients
  late final ApiClient client;
  late final ConnectorApi _connectorApi;
  late final WorkstreamSummaryApi _workstreamSummaryApi;
  late final AnnotationApi _annotationApi;
  late final UserApi _userApi;

  // Application context
  Context? _context;

  // Application info
  final ApplicationNameEnum appName = ApplicationNameEnum.OPEN_SOURCE;
  final String appVersion = "0.0.1";
  final PlatformEnum platform = Platform.operatingSystem == "windows"
      ? PlatformEnum.WINDOWS
      : Platform.operatingSystem == "macos"
      ? PlatformEnum.MACOS
      : PlatformEnum.LINUX;

  // Track if service is initialized
  bool _isInitialized = false;

  /// Placeholder persona string
  /// generation prompts later.
  String persona = '';

  PiecesOSService() {
    client = ApiClient(basePath: baseUrl);
    _connectorApi = ConnectorApi(client);
    _workstreamSummaryApi = WorkstreamSummaryApi(client);
    _annotationApi = AnnotationApi(client);
    _userApi = UserApi(client);
  }

  /// Register/Connect the application to Pieces OS
  Future<Application> connectApplication() async {
    if (_context?.application != null) {
      return _context!.application;
    }

    try {
      final seededApp = SeededTrackedApplication(
        name: appName,
        platform: platform,
        version: appVersion,
      );

      final connection = SeededConnectorConnection(application: seededApp);

      _context = await _connectorApi.connect(
        seededConnectorConnection: connection,
      );

      if (_context?.application == null) {
        throw Exception(
          'Failed to connect to Pieces OS: No application returned',
        );
      }

      dev.log(
        'Successfully connected to Pieces OS: ${_context!.application.name}',
        name: 'PiecesOSService',
      );
      return _context!.application;
    } catch (e, st) {
      dev.log(
        'Error connecting to Pieces OS',
        name: 'PiecesOSService',
        error: e,
        stackTrace: st,
      );
      rethrow;
    }
  }

  /// Initialize the service - connects to Pieces OS.
  ///
  /// NOTE: We intentionally do **not** start any WebSocket listeners. This app
  /// fetches workstream summary identifiers on-demand.
  Future<void> initialize() async {
    if (_isInitialized) {
      dev.log('Service already initialized', name: 'PiecesOSService');
      return;
    }

    try {
      // Connect to Pieces OS
      await connectApplication();

      _isInitialized = true;
      dev.log('Initialized successfully', name: 'PiecesOSService');
    } catch (e, st) {
      dev.log(
        'Error initializing',
        name: 'PiecesOSService',
        error: e,
        stackTrace: st,
      );
      rethrow;
    }
  }

Step 3: retrieve recent work context (workstream summaries)

To ground generation in “what I actually worked on”, the service does this on-demand:

fetchLatestWorkstreamSummaryIds() opens a WebSocket to the identifiers stream, reads one payload, then closes the socket.
getLastSummaryContents() uses those IDs to fetch summary snapshots and extracts the DESCRIPTION annotation text via getSummaryContent().

Those DESCRIPTION strings are what we feed into the LLM as “recent context”.

Here’s the exact code for streaming IDs, fetching summaries, and extracting DESCRIPTION text:

  /// Fetch the most recent workstream summary identifiers on-demand.
  ///
  /// This opens a WebSocket connection **once**, reads the first identifiers
  /// payload, then closes the socket. No continuous listener / no caching.
  Future<List<String>> fetchLatestWorkstreamSummaryIds({
    int limit = 10,
    Duration timeout = const Duration(seconds: 8),
  }) async {
    if (limit <= 0) return const [];
    await initialize();

    final wsUrl = '$websocketUrl/workstream_summaries/stream/identifiers';
    WebSocket? socket;
    try {
      socket = await WebSocket.connect(wsUrl).timeout(timeout);
      final first = await socket.first.timeout(timeout);

      String raw;
      if (first is String) {
        raw = first;
      } else if (first is List<int>) {
        raw = utf8.decode(first);
      } else {
        raw = first.toString();
      }

      final decoded = jsonDecode(raw);
      final streamed = StreamedIdentifiers.fromJson(decoded);
      final ids = (streamed?.iterable ?? const [])
          .map((e) => e.workstreamSummary?.id)
          .whereType<String>()
          .where((s) => s.trim().isNotEmpty)
          .take(limit)
          .toList(growable: false);
      return ids;
    } finally {
      try {
        await socket?.close();
      } catch (_) {
        // ignore
      }
    }
  }

  /// Get the summary content from a workstream summary's annotations
  Future<String?> getSummaryContent(WorkstreamSummary summary) async {
    try {
      // Loop through annotations to find the DESCRIPTION type
      for (final annotationRef
          in summary.annotations?.indices.keys.toList() ?? []) {
        // Fetch the full annotation using AnnotationApi (singular)
        final annotation = await _annotationApi
            .annotationSpecificAnnotationSnapshot(annotationRef);
        if (annotation == null) {
          continue;
        }

        // Check if this is a DESCRIPTION type annotation
        if (annotation.type == AnnotationTypeEnum.DESCRIPTION) {
          // Return the text content
          return annotation.text;
        }
      }

      return null;
    } catch (e) {
      dev.log(
        'Error fetching annotation content for ${summary.id}: $e',
        name: 'PiecesOSService',
      );
      return null;
    }
  }

  /// Get the last [limit] workstream summary DESCRIPTION texts (most recent first).
  ///
  /// This performs an on-demand identifiers fetch, then retrieves each summary
  /// snapshot and its DESCRIPTION annotation text.
  Future<List<String>> getLastSummaryContents({int limit = 10}) async {
    if (limit <= 0) return const [];
    await initialize();

    final ids = await fetchLatestWorkstreamSummaryIds(limit: limit);
    if (ids.isEmpty) return const [];

    final summaries = await Future.wait(
      ids.map(
        (id) => _workstreamSummaryApi
            .workstreamSummariesSpecificWorkstreamSummarySnapshot(id),
      ),
    );

    final contents = await Future.wait(
      summaries.whereType<WorkstreamSummary>().map(
        (s) async => (await getSummaryContent(s))?.trim(),
      ),
    );

    return contents.whereType<String>().where((t) => t.isNotEmpty).toList();
  }

Step 4: retrieve persona signals (user annotations)

To personalize voice and framing, getPersonaFromUserAnnotations():

resolves the active user (_resolveUserId() → UserApi.userSnapshot())
fetches recent annotations (UserApi.userGetAnnotations(...))
normalizes them into a prompt-friendly PersonaAnnotations object

If Pieces returns no annotations, we keep persona optional (so the model doesn’t invent one).

Here’s the code that resolves the active user and turns annotations into prompt-friendly “persona signals”:

  Future<String> _resolveUserId() async {
    await initialize();

    final snap = await _userApi.userSnapshot();
    final userId = snap?.user?.id;
    if (userId != null && userId.trim().isNotEmpty) return userId.trim();

    throw StateError(
      'No active user found. Ensure Pieces has an active user session.',
    );
  }

  /// Get a "persona" text derived from the user's annotations.
  ///
  /// Uses `UserApi.userGetAnnotations()` which returns the resolved active Person
  /// and filtered Annotations.
  Future<PersonaAnnotations> getPersonaFromUserAnnotations({
    int limit = 1,
  }) async {
    if (limit <= 0) return const PersonaAnnotations();
    final userId = await _resolveUserId();
    final out = await _userApi.userGetAnnotations(
      userId,
      UserAnnotationsInput(limit: limit),
    );

    final texts = out.annotations.iterable
        .map((a) => a.text)
        .whereType<String>()
        .map((t) => t.trim())
        .where((t) => t.isNotEmpty)
        .toList();
    dev.log(
      'Fetched ${texts.length} annotation texts for persona.',
      name: 'PiecesOSService',
    );

    // Pieces might return an empty annotations list; in that case we return empty
    // and let the caller omit persona entirely.
    if (texts.isEmpty) return const PersonaAnnotations();

    final normalized = <String>[];
    for (final t in texts) {
      final oneLine = t.replaceAll(RegExp(r'\s+'), ' ').trim();
      if (oneLine.isEmpty) continue;
      normalized.add(
        oneLine.length > 180 ? '${oneLine.substring(0, 180)}…' : oneLine,
      );
    }

    return PersonaAnnotations(annotations: normalized);
  }

Step 5: enable RAG with MCP (agentic mode)

When we connect via the MCP endpoint, the generator can do RAG (retrieval‑augmented generation):

connectMcp() connects a streamable HTTP transport (POST + SSE GET)
listTools() fetches and caches the available MCP tools
callTool() lets the agent retrieve/verify relevant context from Pieces during planning/writing

This is the “agentic” part: instead of guessing, the model can retrieve what it needs.

Here’s the MCP client code that enables tool calls (MCP endpoint → RAG):

  // MCP (Streamable HTTP/SSE) client
  mcp.McpClient? _mcpClient;
  mcp.StreamableHttpClientTransport? _mcpTransport;
  Uri? _mcpEndpoint;
  List<mcp.Tool>? _cachedMcpTools;

  bool get isMcpConnected =>
      _mcpClient != null && _mcpTransport != null;

  /// Connect to Pieces MCP endpoint using Streamable HTTP (POST + SSE GET).
  ///
  /// Default endpoint is `http://localhost:39300/mcp`.
  Future<void> connectMcp({String? endpoint}) async {
    final ep = Uri.parse((endpoint ?? defaultMcpEndpoint).trim());

    // If already connected to same endpoint, do nothing.
    if (isMcpConnected && _mcpEndpoint == ep) return;

    // Close any existing transport/client.
    await disconnectMcp();
    _mcpEndpoint = ep;

    final transport = mcp.StreamableHttpClientTransport(ep);
    final client = mcp.McpClient(
      const mcp.Implementation(name: 'blog_generator', version: '0.0.1'),
      options: const mcp.McpClientOptions(
        // Client-side capabilities are for server-initiated requests
        // (sampling, elicitation, tasks, roots). We don't need any for now.
        capabilities: mcp.ClientCapabilities(),
      ),
    );

    transport.onerror = (err) {
      dev.log(
        'MCP transport error: $err',
        name: 'PiecesOSService',
        error: err,
      );
    };
    transport.onclose = () {
      dev.log('MCP transport closed', name: 'PiecesOSService');
      _cachedMcpTools = null;
      _mcpClient = null;
      _mcpTransport = null;
    };

    try {
      await client.connect(transport);
      _mcpTransport = transport;
      _mcpClient = client;
      final server = client.getServerVersion();
      dev.log(
        'MCP connected: ${server?.name ?? 'unknown'} ${server?.version ?? ''}',
        name: 'PiecesOSService',
      );
    } catch (e) {
      await disconnectMcp();
      rethrow;
    }
  }

  Future<void> disconnectMcp() async {
    _cachedMcpTools = null;
    try {
      await _mcpTransport?.terminateSession();
    } catch (_) {
      // Ignore - server may not support.
    }

    try {
      await _mcpClient?.close();
    } catch (_) {
      // Ignore.
    }
    try {
      await _mcpTransport?.close();
    } catch (_) {
      // Ignore.
    }

    _mcpClient = null;
    _mcpTransport = null;
  }

  Future<List<mcp.Tool>> listTools({bool forceRefresh = false}) async {
    final client = _mcpClient;
    if (client == null) {
      throw StateError('MCP client not connected');
    }

    if (!forceRefresh && _cachedMcpTools != null) return _cachedMcpTools!;

    final res = await client.listTools();
    _cachedMcpTools = res.tools;
    return res.tools;
  }

  Future<mcp.CallToolResult> callTool({
    required String name,
    Map<String, dynamic> arguments = const {},
  }) async {
    final client = _mcpClient;
    if (client == null) {
      throw StateError('MCP client not connected');
    }

    return client.callTool(
      mcp.CallToolRequest(name: name, arguments: arguments),
    );
  }

Step 6: clean up

dispose() closes the MCP session (best-effort) and clears state so the service doesn’t leak resources across UI lifecycles.

Here’s the cleanup method from the service (it also closes the class):

  /// Disconnect from Pieces OS and cleanup resources
  void dispose() {
    _isInitialized = false;

    // Close MCP client/transport
    // Fire and forget; dispose is sync.
    unawaited(disconnectMcp());

    // Clear context
    _context = null;

    dev.log('Disposed', name: 'PiecesOSService');
  }
}

How the UI uses this service (quick walkthrough)

In BlogWizardScreen, the flow looks like this:

Get grounded context
- _pieces.getLastSummaryContents(limit: 10) for recent workstream summaries
- _pieces.getPersonaFromUserAnnotations(limit: 1) for persona signals (optional)
Turn on agentic RAG
- _pieces.connectMcp(...) then _pieces.listTools()
- pass callMcpTool: (...) => _pieces.callTool(...) into the agent loop

From there, the generator improves quality by moving from one-shot output to a structured workflow:

plan part titles first
generate outlines per part (reviewable)
write each part with the outline as a contract

Takeaway

Pieces OS is what makes this blog generator feel real:

it anchors generation in your actual recent work (summaries)
it shapes tone/voice via your own signals (annotations → persona)
it enables agentic correctness when available (MCP tools)

And the prompt strategy matters just as much: moving from one-shot generation to a plan-first, titled multi-part workflow is what consistently turns “okay output” into “publishable output”.

The rest of the project code (UI, models, generation logic, widgets, etc.) is available on GitHub repository.

Building a TUI with Pieces SDK - Part 3: Advanced Features

Pieces 🌟 — Mon, 02 Feb 2026 17:57:24 +0000

Building a Pieces Copilot TUI - Part 3: Advanced Features & Integration

Note: This tutorial is part of the Pieces CLI project. We welcome contributions! Feel free to open issues, submit PRs, or suggest improvements.

Introduction

Welcome to Part 3! In Part 2, we built the core UI components. Now, we'll add the advanced features to create a fully-functional Pieces Copilot TUI.

What we built in Part 2:

✅ Message widget (chat_message.py)
✅ Chat panel (chat_panel.py)
✅ Input widget (chat_input.py)
✅ Basic view (chat_view.py - simple version)
✅ Main application (app.py)

Step 1: Create the Conversations List

Let's build chats_list.py - the sidebar showing all conversations:

# chats_list.py
"""Conversations list panel widget."""

from typing import Optional
from textual.widgets import Static, Button
from textual.containers import VerticalScroll, Vertical
from textual.message import Message

from pieces_os_client import PiecesClient
from pieces_os_client.wrapper.basic_identifier.chat import BasicChat


class ChatSelected(Message):
    """Message emitted when a chat is selected."""

    def __init__(self, chat: BasicChat):
        super().__init__()
        self.chat = chat


class NewChatRequested(Message):
    """Message emitted when new chat is requested."""
    pass


class ChatItem(Static):
    """Widget representing a single chat in the list."""

    DEFAULT_CSS = """
    ChatItem {
        width: 100%;
        height: auto;
        padding: 1 2;
        margin-bottom: 1;
        background: $surface;
        border-left: solid $primary;
    }

    ChatItem:hover {
        background: $panel;
        border-left: solid $accent;
    }

    ChatItem.active {
        background: $primary 30%;
        border-left: thick $accent;
        text-style: bold;
    }

    ChatItem .chat-title {
        color: $text;
        text-style: bold;
    }

    ChatItem .chat-summary {
        color: $text-muted;
        text-style: italic;
    }
    """

    def __init__(self, chat: BasicChat, **kwargs):
        super().__init__(**kwargs)
        self.chat = chat

    def compose(self):
        """Compose the chat item."""
        title = self.chat.name or "Untitled"
        summary = self.chat.summary or ""
        if len(summary) > 50:
            summary = summary[:47] + "..."

        yield Static(f"💬 {title}", classes="chat-title")
        if summary:
            yield Static(summary, classes="chat-summary")

    def on_click(self) -> None:
        """Handle click event."""
        self.post_message(ChatSelected(self.chat))


class ChatsList(Vertical):
    """Panel to display list of conversations."""

    DEFAULT_CSS = """
    ChatsList {
        width: 25%;
        height: 100%;
        border: solid $primary;
        background: $background;
        padding: 1;
    }

    ChatsList:focus-within {
        border: solid $accent;
    }

    ChatsList Button {
        width: 100%;
        margin-bottom: 1;
    }

    ChatsList VerticalScroll {
        height: 1fr;
    }

    ChatsList .empty-state {
        text-align: center;
        color: $text-muted;
        text-style: italic;
        margin: 2;
    }
    """

    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.border_title = "Conversations"
        self.active_chat: Optional[BasicChat] = None
        self._chat_items = {}

    def compose(self):
        """Compose the chats list panel."""
        yield Button("➕ New Chat", id="new-chat-btn")
        yield VerticalScroll(id="chats-container")

    def load_chats(self, pieces_client: PiecesClient):
        """Load chats from the API."""
        try:
            chats = pieces_client.copilot.chats()

            container = self.query_one("#chats-container", VerticalScroll)

            # Clear existing items
            container.remove_children()
            self._chat_items.clear()

            if not chats:
                container.mount(Static("No chats yet...", classes="empty-state"))
            else:
                for chat in chats:
                    chat_item = ChatItem(chat)
                    self._chat_items[chat.id] = chat_item
                    container.mount(chat_item)
        except (ConnectionError, AttributeError) as e:
            container = self.query_one("#chats-container", VerticalScroll)
            container.mount(Static(f"❌ Failed to load chats: {str(e)}", classes="empty-state"))

    def set_active_chat(self, chat: Optional[BasicChat]):
        """Set the active chat and update UI."""
        # Remove active class from all items
        for item in self._chat_items.values():
            item.remove_class("active")

        self.active_chat = chat

        # Add active class to the selected item
        if chat and chat.id in self._chat_items:
            self._chat_items[chat.id].add_class("active")

    def add_new_chat(self, chat: BasicChat):
        """Add a new chat to the list."""
        container = self.query_one("#chats-container", VerticalScroll)

        # Remove empty state if present
        try:
            empty_state = container.query_one(".empty-state")
            empty_state.remove()
        except LookupError:
            # No empty state widget present, which is fine
            pass

        # Add new chat at the top
        chat_item = ChatItem(chat)
        self._chat_items[chat.id] = chat_item

        # Mount at the beginning
        if container.children:
            container.mount(chat_item, before=container.children[0])
        else:
            container.mount(chat_item)

    def on_button_pressed(self, event: Button.Pressed) -> None:
        """Handle button press."""
        if event.button.id == "new-chat-btn":
            self.post_message(NewChatRequested())

Key features:

ChatItem - Individual conversation with title and summary
ChatSelected - Custom message when chat is clicked
NewChatRequested - Message when "New Chat" button is pressed
Active chat highlighting
Hover effects
Error handling for API failures

Step 2: Create the Streaming Handler

Now src/pieces_copilot_tui/streaming_handler.py - connects Pieces OS streaming to our UI:

# src/pieces_copilot_tui/streaming_handler.py
"""Handler for streaming responses from Pieces OS."""

from typing import Callable, Optional

from pieces_os_client.wrapper import PiecesClient
from pieces_os_client.wrapper.basic_identifier.chat import BasicChat


class StreamingHandler:
    """Handles streaming responses from the Pieces Copilot."""

    def __init__(
        self,
        pieces_client: PiecesClient,
        on_thinking_started: Callable[[], None],
        on_stream_started: Callable[[str], None],
        on_stream_chunk: Callable[[str], None],
        on_stream_completed: Callable[[Optional[BasicChat]], None],
        on_stream_error: Callable[[str], None],
    ):
        """
        Initialize the streaming handler.

        Args:
            pieces_client: The Pieces client instance
            on_thinking_started: Callback when thinking starts
            on_stream_started: Callback when streaming starts (with initial text)
            on_stream_chunk: Callback for each chunk (with full accumulated text)
            on_stream_completed: Callback when streaming completes (with optional new chat)
            on_stream_error: Callback when an error occurs (with error message)
        """
        self.pieces_client = pieces_client
        self.on_thinking_started = on_thinking_started
        self.on_stream_started = on_stream_started
        self.on_stream_chunk = on_stream_chunk
        self.on_stream_completed = on_stream_completed
        self.on_stream_error = on_stream_error

        self._current_response = ""
        self._current_status: Optional[str] = None

        # Register callback
        if self.pieces_client.copilot.ask_stream_ws:
            self.pieces_client.copilot.ask_stream_ws.on_message_callback = (
                self._handle_stream_message
            )

    def ask_question(self, query: str) -> None:
        """
        Ask a question to the copilot.

        Args:
            query: The question to ask
        """
        self._current_response = ""
        self._current_status = None
        self.on_thinking_started()
        self.pieces_client.copilot.stream_question(query)

    def _handle_stream_message(self, response) -> None:
        """Handle streaming messages from the copilot."""
        try:
            status = response.status

            if status == "IN-PROGRESS":
                if response.question:
                    answers = response.question.answers.iterable
                    for answer in answers:
                        if answer.text:
                            if not self._current_response:
                                # First chunk - start streaming
                                self._current_response = answer.text
                                self.on_stream_started(self._current_response)
                            else:
                                # Subsequent chunks
                                self._current_response += answer.text
                                self.on_stream_chunk(self._current_response)

            elif status == "COMPLETED":
                # Streaming completed
                new_chat = None
                if response.conversation:
                    new_chat = BasicChat(response.conversation)
                    self.pieces_client.copilot.chat = new_chat

                self.on_stream_completed(new_chat)
                self._current_response = ""

            elif status in ["FAILED", "STOPPED", "CANCELED"]:
                # Handle error
                error_msg = getattr(response, "error_message", "Unknown error")
                self.on_stream_error(error_msg)
                self._current_response = ""

        except (AttributeError, ConnectionError, ValueError) as e:
            self.on_stream_error(str(e))
            self._current_response = ""

Step 3: Create the Full View Orchestrator

Now the complete chat_view.py - replaces the simple version from Part 2:

# chat_view.py
"""Main copilot view combining all widgets."""

from typing import Optional
from textual.screen import Screen
from textual.containers import Horizontal, Vertical
from textual.binding import Binding
from textual.widgets import Footer

from pieces_os_client import PiecesClient
from pieces_os_client.wrapper.basic_identifier.chat import BasicChat

from .chat_panel import ChatPanel
from .chats_list import ChatsList, ChatSelected, NewChatRequested
from .chat_input import ChatInput, MessageSubmitted
from .streaming_handler import StreamingHandler


class CopilotView(Screen):
    """Main copilot view screen."""

    BINDINGS = [
        Binding("ctrl+n", "new_chat", "New Chat"),
        Binding("ctrl+r", "rename_chat", "Rename"),
        Binding("ctrl+d", "delete_chat", "Delete"),
        Binding("ctrl+l", "toggle_ltm", "Toggle LTM"),
        Binding("ctrl+q", "quit", "Quit"),
    ]

    DEFAULT_CSS = """
    CopilotView {
        layout: vertical;
    }

    CopilotView Horizontal {
        height: 1fr;
    }

    CopilotView Vertical.main-content {
        width: 75%;
        layout: vertical;
    }
    """

    def __init__(self, pieces_client: PiecesClient, **kwargs):
        super().__init__(**kwargs)
        self.pieces_client = pieces_client
        self.current_chat: Optional[BasicChat] = None
        self.chat_panel: Optional[ChatPanel] = None
        self.chats_list: Optional[ChatsList] = None
        self.chat_input: Optional[ChatInput] = None
        self._ltm_enabled = False
        self.streaming_handler: Optional[StreamingHandler] = None

    def compose(self):
        """Compose the view."""
        with Horizontal():
            # Left sidebar - conversations list
            self.chats_list = ChatsList()
            yield self.chats_list

            # Right side - chat panel and input
            with Vertical(classes="main-content"):
                self.chat_panel = ChatPanel()
                yield self.chat_panel

                self.chat_input = ChatInput()
                yield self.chat_input

        yield Footer()

    def on_mount(self) -> None:
        """Initialize the view."""
        # Load chats
        self.chats_list.load_chats(self.pieces_client)

        # Show welcome message
        self.chat_panel.show_welcome()

        # Check LTM status
        try:
            self._ltm_enabled = self.pieces_client.copilot.is_chat_ltm_enabled()
        except (ConnectionError, AttributeError):
            self._ltm_enabled = False

        # Setup streaming handler
        self.streaming_handler = StreamingHandler(
            pieces_client=self.pieces_client,
            on_thinking_started=self._on_thinking_started,
            on_stream_started=self._on_stream_started,
            on_stream_chunk=self._on_stream_chunk,
            on_stream_completed=self._on_stream_completed,
            on_stream_error=self._on_stream_error,
        )

    def on_chat_selected(self, message: ChatSelected) -> None:
        """Handle chat selection."""
        self.current_chat = message.chat
        self.pieces_client.copilot.chat = message.chat
        self.chats_list.set_active_chat(message.chat)

        # Load conversation
        self.chat_panel.clear_messages()
        self.chat_panel.border_title = f"Chat: {message.chat.name}"

        try:
            messages = message.chat.messages()
            for msg in messages:
                role = msg.role
                content = msg.raw_content
                self.chat_panel.add_message(role, content)
        except (ConnectionError, AttributeError, ValueError) as e:
            self.chat_panel.add_message("system", f"❌ Error loading messages: {str(e)}")

    def on_new_chat_requested(self, _: NewChatRequested) -> None:
        """Handle new chat request."""
        self.action_new_chat()

    def on_message_submitted(self, message: MessageSubmitted) -> None:
        """Handle user message submission."""
        if self.chat_panel.is_streaming_active():
            return

        # Add user message to chat panel
        self.chat_panel.add_message("user", message.text)

        # Send to copilot via streaming handler
        if self.streaming_handler:
            self.streaming_handler.ask_question(message.text)

    # Streaming handler callbacks
    def _on_thinking_started(self) -> None:
        """Called when copilot starts thinking."""
        self.call_from_thread(self.chat_panel.add_thinking_indicator)

    def _on_stream_started(self, initial_text: str) -> None:
        """Called when streaming starts with initial text."""
        self.call_from_thread(
            self.chat_panel.add_streaming_message,
            "assistant",
            initial_text
        )

    def _on_stream_chunk(self, full_text: str) -> None:
        """Called for each streaming chunk with accumulated text."""
        self.call_from_thread(
            self.chat_panel.update_streaming_message,
            full_text
        )

    def _on_stream_completed(self, new_chat: Optional[BasicChat]) -> None:
        """Called when streaming completes."""
        self.call_from_thread(self.chat_panel.finalize_streaming_message)

        # Update chat reference if new chat was created
        if new_chat:
            old_chat = self.current_chat
            if not old_chat or old_chat.id != new_chat.id:
                self.current_chat = new_chat
                self.call_from_thread(self.chats_list.add_new_chat, new_chat)
                self.call_from_thread(self.chats_list.set_active_chat, new_chat)
                self.chat_panel.border_title = f"Chat: {new_chat.name}"

    def _on_stream_error(self, error_msg: str) -> None:
        """Called when a streaming error occurs."""
        self.call_from_thread(
            self.chat_panel.add_message,
            "system",
            f"❌ Error: {error_msg}"
        )

    def action_new_chat(self):
        """Create a new chat."""
        self.current_chat = None
        self.pieces_client.copilot.chat = None
        self.chats_list.set_active_chat(None)
        self.chat_panel.clear_messages()
        self.chat_panel.border_title = "Chat: New Conversation"
        self.chat_panel.show_welcome()
        self.chat_input.focus()

    def action_rename_chat(self):
        """Rename the current chat."""
        if not self.current_chat:
            self.notify("No chat selected", severity="warning")
            return

        # For now, just show a notification - you can implement a dialog later
        self.notify("Rename chat - Not implemented yet", severity="info")

    def action_delete_chat(self):
        """Delete the current chat."""
        if not self.current_chat:
            self.notify("No chat selected", severity="warning")
            return

        try:
            chat_name = self.current_chat.name
            self.current_chat.delete()
            self.notify(f"Deleted chat: {chat_name}", severity="success")

            # Clear view and reload chats
            self.action_new_chat()
            self.chats_list.load_chats(self.pieces_client)

        except (ConnectionError, AttributeError) as e:
            self.notify(f"Error deleting chat: {str(e)}", severity="error")

    def action_toggle_ltm(self):
        """Toggle LTM (Long Term Memory)."""
        try:
            is_chat_ltm_enabled = self.pieces_client.copilot.is_chat_ltm_enabled()

            if is_chat_ltm_enabled:
                self.pieces_client.copilot.deactivate_ltm()
                self.notify("🧠 Chat LTM disabled", severity="info")
                self._ltm_enabled = False
            else:
                is_system_ltm_running = self.pieces_client.copilot.is_ltm_running()

                if is_system_ltm_running:
                    self.pieces_client.copilot.activate_ltm()
                    self.notify("🧠 Chat LTM enabled", severity="success")
                    self._ltm_enabled = True
                else:
                    self.notify("🧠 LTM system not running. Please enable it first.", severity="warning")

        except (ConnectionError, AttributeError) as e:
            self.notify(f"❌ Error toggling LTM: {str(e)}", severity="error")

🚨 Critical for Thread Safety

Always use call_from_thread() when updating UI from background threads. Direct UI updates from threads will cause crashes! The streaming handler runs in a background thread, so all UI updates must go through this method.

Key features:

Split-pane layout (25% sidebar, 75% chat)
Handles all custom messages (ChatSelected, NewChatRequested, MessageSubmitted)
Thread-safe UI updates via call_from_thread()
Keyboard shortcuts (Ctrl+N, Ctrl+D, Ctrl+L, Ctrl+Q)
Conversation management
LTM toggle support

Step 4: Update the Main Application

Now update src/pieces_copilot_tui/app.py to use the full CopilotView:

# src/pieces_copilot_tui/app.py
"""Pieces Copilot TUI - A Terminal User Interface for Pieces OS."""

from textual.app import App

from pieces_os_client.wrapper import PiecesClient
from .chat_view import CopilotView


class PiecesCopilotTUI(App):
    """Pieces Copilot TUI with full features."""

    # CSS styles for the entire app
    DEFAULT_CSS = """
    Screen {
        background: $background;
        color: $text;
    }

    .error {
        color: $error;
        text-style: bold;
    }

    .success {
        color: $success;
        text-style: bold;
    }

    .warning {
        color: $warning;
        text-style: bold;
    }
    """

    def __init__(self, pieces_client: PiecesClient = None, **kwargs):
        super().__init__(**kwargs)
        self.pieces_client = pieces_client or PiecesClient()
        self.copilot_view = None

    def on_mount(self) -> None:
        """Initialize the application."""
        self.title = "Pieces Copilot TUI"

        # Initialize Pieces client
        if not self.pieces_client.connect_websocket():
            self.notify("Failed to connect to Pieces OS", severity="error")
            self.exit()
            return

        # Create and push the full copilot view
        self.copilot_view = CopilotView(pieces_client=self.pieces_client)
        self.push_screen(self.copilot_view)


def run_tui():
    """Run the Pieces Copilot TUI application."""
    # Initialize Pieces client
    pieces_client = PiecesClient()

    # Check if Pieces OS is running
    if not pieces_client.is_pieces_running():
        print("Error: Pieces OS is not running. Please start Pieces OS first.")
        return

    app = PiecesCopilotTUI(pieces_client=pieces_client)
    app.run()


if __name__ == "__main__":
    run_tui()

Changes from Part 2:

Import CopilotView instead of SimpleChatView
Added warning CSS class
Now using the full-featured view

Step 5: Run Your Complete TUI

The __main__.py file is already set up from Part 2. Run it the Pythonic way:

# Navigate to src directory
cd src

# Run the module
python -m pieces_copilot_tui

That's it! Simple, clean, and Pythonic. ✨

🎉 You now have a fully functional Pieces Copilot TUI!

Add Debug Logging

# In any widget method
self.log(f"Current state: {some_variable}")
self.log.info("Info message")
self.log.warning("Warning message")
self.log.error("Error message")

# View in textual console

Common Issues

Problem: Streaming doesn't start

# Check WebSocket connection
if not client.copilot.ask_stream_ws:
    print("❌ WebSocket not connected!")
    client.connect_websocket()

Problem: UI not updating from streaming

# Make sure you're using call_from_thread
self.call_from_thread(self.chat_panel.add_message, "assistant", text)
# NOT: self.chat_panel.add_message("assistant", text)

Problem: Messages not loading

# Add error handling
try:
    messages = chat.messages()
except Exception as e:
    self.log.error(f"Failed to load messages: {e}")

Building a TUI with Pieces SDK - Part 2: UI Components

Pieces 🌟 — Mon, 02 Feb 2026 17:57:15 +0000

Building a Pieces Copilot TUI - Part 2: Basic UI Components

Note: This tutorial is part of the Pieces CLI project. We welcome contributions! Feel free to open issues, submit PRs, or suggest improvements.

Introduction

Welcome to Part 2! In Part 1, we learned how to use the Pieces OS SDK. Now, we'll start building the Terminal User Interface (TUI) using Textual.

What we'll build in Part 2:

Understanding Textual fundamentals
Message display widget
Chat panel for conversations
User input widget
A working basic chat interface

What we'll add in Part 3:

Conversations list sidebar
Streaming handler integration
Full view orchestration
Testing & optimization

Understanding Textual

Textual is a modern Python framework for building TUIs. It uses a reactive, component-based architecture similar to React.

Project Structure

pieces-copilot-tui/
├── venv/                                    # Virtual environment
├── requirements.txt                         # Dependencies
├── run.py                                   # Entry point
└── src/                                     # Source code
    └── pieces_copilot_tui/                  # Main package
        ├── __init__.py                      # Package initialization
        ├── app.py                           # Main TUI application
        ├── chat_message.py                  # Individual message widget (Part 2)
        ├── chat_panel.py                    # Message display panel (Part 2)
        ├── chat_input.py                    # User input widget (Part 2)
        ├── streaming_handler.py             # Pieces SDK streaming logic (Part 3)
        ├── chats_list.py                    # Conversations list (Part 3)
        └── chat_view.py                     # Main view orchestrator (Part 3)

In Part 2, we'll build the core UI components. In Part 3, we'll add conversations management and full integration.

Step 1: Create the Message Widget

Let's start with the foundation - src/pieces_copilot_tui/chat_message.py:

# src/pieces_copilot_tui/chat_message.py
"""Chat message widget for displaying individual messages."""

from datetime import datetime
from textual.widgets import Static
from textual.containers import Vertical


class ChatMessage(Vertical):
    """Widget to display a single chat message."""

    DEFAULT_CSS = """
    ChatMessage {
        width: 100%;
        height: auto;
        padding: 1 2;
        margin-bottom: 1;
    }

    ChatMessage.user-message {
        background: $primary 20%;
        border-left: thick $primary;
    }

    ChatMessage.assistant-message {
        background: $accent 20%;
        border-left: thick $accent;
    }

    ChatMessage.system-message {
        background: $warning 20%;
        border-left: thick $warning;
    }

    ChatMessage .message-header {
        color: $text-muted;
        text-style: italic;
        margin-bottom: 1;
    }

    ChatMessage .message-content {
        color: $text;
    }
    """

    def __init__(self, role: str, content: str, timestamp: str = None, **kwargs):
        super().__init__(**kwargs)
        self.role = role
        self.content = content
        self.timestamp = timestamp or datetime.now().strftime("Today %I:%M %p")

        # Set CSS class based on role
        self.add_class(f"{role}-message")

    def compose(self):
        """Compose the message widget."""
        role_emoji = {
            "user": "👤",
            "assistant": "🤖",
            "system": "⚙️"
        }
        emoji = role_emoji.get(self.role, "")

        yield Static(
            f"{emoji} {self.role.title()} - {self.timestamp}",
            classes="message-header"
        )
        yield Static(self.content, classes="message-content")

Key concepts:

Vertical container stacks children vertically
compose() defines child widgets
CSS classes are added dynamically based on message role
Emojis make it visually appealing! 🎨

CSS Variables:

$primary, $accent, $warning - Textual's built-in color variables
$text, $text-muted - Text color variables
You can customize these in your app's theme

Step 2: Create the Chat Panel

Now src/pieces_copilot_tui/chat_panel.py - the scrollable message display:

# src/pieces_copilot_tui/chat_panel.py
"""Chat panel widget for displaying conversation messages."""

from datetime import datetime
from typing import Optional
from textual.widgets import Static
from textual.containers import VerticalScroll

from .chat_message import ChatMessage


class ChatPanel(VerticalScroll):
    """Panel to display chat messages."""

    DEFAULT_CSS = """
    ChatPanel {
        width: 100%;
        height: 1fr;
        border: solid $primary;
        background: $background;
        padding: 1 2;
    }

    ChatPanel:focus {
        border: solid $accent;
    }

    ChatPanel .thinking-indicator {
        color: $warning;
        text-style: italic bold blink;
        text-align: center;
        background: $surface;
        border: dashed $warning;
        padding: 1;
        margin: 1;
    }

    ChatPanel .streaming-message {
        border-left: thick $accent;
        text-style: bold;
    }

    ChatPanel .welcome-message {
        text-align: center;
        margin: 4 2;
        padding: 3;
        border: dashed $primary;
        background: $primary 10%;
        color: $text;
        width: 100%;
        height: auto;
    }
    """

    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.border_title = "Chat"
        self._messages = []
        self._thinking_widget = None
        self._streaming_widget = None

    def add_message(self, role: str, content: str, timestamp: str = None):
        """Add a new message to the chat panel."""
        self._clear_thinking_indicator()

        timestamp = timestamp or datetime.now().strftime("Today %I:%M %p")
        message = ChatMessage(role, content, timestamp)
        self._messages.append(message)
        self.mount(message)
        self.scroll_end(animate=False)

    def add_thinking_indicator(self):
        """Add a thinking indicator."""
        self._clear_thinking_indicator()
        self._thinking_widget = Static("🤔 Thinking...", classes="thinking-indicator")
        self.mount(self._thinking_widget)
        self.scroll_end(animate=False)

    def add_streaming_message(self, role: str, content: str):
        """Add a streaming message with cursor."""
        self._clear_thinking_indicator()

        timestamp = datetime.now().strftime("Today %I:%M %p")
        self._streaming_widget = ChatMessage(role, content + " ▌", timestamp)
        self._streaming_widget.add_class("streaming-message")
        self.mount(self._streaming_widget)
        self.scroll_end(animate=False)

    def update_streaming_message(self, content: str):
        """Update the streaming message content."""
        if self._streaming_widget:
            # Update the content of the last Static widget in the streaming message
            content_widget = self._streaming_widget.query(".message-content").first()
            if content_widget:
                content_widget.update(content + " ▌")
            self.scroll_end(animate=False)

    def finalize_streaming_message(self):
        """Convert streaming message to permanent message."""
        if self._streaming_widget:
            content_widget = self._streaming_widget.query(".message-content").first()
            if content_widget:
                # Remove cursor
                current_content = str(content_widget.renderable).replace(" ▌", "")
                content_widget.update(current_content)
            self._streaming_widget.remove_class("streaming-message")
            self._messages.append(self._streaming_widget)
            self._streaming_widget = None

    def clear_messages(self):
        """Clear all messages from the chat panel."""
        self._clear_thinking_indicator()
        if self._streaming_widget:
            self._streaming_widget.remove()
            self._streaming_widget = None

        for message in self._messages:
            if message.is_mounted:
                message.remove()
        self._messages.clear()

    def show_welcome(self):
        """Show welcome message."""
        welcome_text = """🎯 Pieces Copilot TUI

Type your message below to start chatting!

Press Ctrl+Q to quit."""

        welcome = Static(welcome_text, classes="welcome-message")
        self.mount(welcome)

    def _clear_thinking_indicator(self):
        """Remove thinking indicator if present."""
        if self._thinking_widget:
            if self._thinking_widget.is_mounted:
                self._thinking_widget.remove()
            self._thinking_widget = None

    def is_streaming_active(self) -> bool:
        """Check if streaming or thinking is currently active."""
        return self._streaming_widget is not None or self._thinking_widget is not None

Key features:

VerticalScroll makes it scrollable
Thinking indicator shows "🤔 Thinking..."
Streaming shows content + cursor (" ▌")
Auto-scrolls to bottom on new messages
Welcome message for first-time users

Step 3: Create the Input Widget

Simple but crucial - src/pieces_copilot_tui/chat_input.py:

# src/pieces_copilot_tui/chat_input.py
"""Chat input widget for user message entry."""

from textual.widgets import Input
from textual.message import Message


class MessageSubmitted(Message):
    """Message emitted when user submits input."""

    def __init__(self, text: str):
        super().__init__()
        self.text = text


class ChatInput(Input):
    """Input widget for chat messages."""

    DEFAULT_CSS = """
    ChatInput {
        width: 100%;
        height: 3;
        background: $surface;
        border: solid $primary;
        padding: 0 1;
        margin: 0;
        dock: bottom;
    }

    ChatInput:focus {
        border: solid $accent;
        background: $panel;
    }
    """

    def __init__(self, **kwargs):
        super().__init__(placeholder="Type your message here...", **kwargs)

    async def on_input_submitted(self, event: Input.Submitted) -> None:
        """Handle input submission."""
        if event.value.strip():
            self.post_message(MessageSubmitted(event.value.strip()))
            self.value = ""
            event.stop()

    def on_mount(self) -> None:
        """Focus the input when mounted."""
        self.focus()

Textual messages:

Custom MessageSubmitted message bubbles up to parent
Parent widgets can listen for this message
event.stop() prevents further propagation
post_message() sends the message up the widget tree

How it works:

User types message and presses Enter
on_input_submitted() is called
Creates MessageSubmitted message with the text
Parent widget receives it via on_message_submitted()
Input is cleared for next message

Step 4: Create a Simple View (For Testing)

Let's create a basic src/pieces_copilot_tui/chat_view.py to test our components:

# src/pieces_copilot_tui/chat_view.py
"""Simple chat view for testing (will be expanded in Part 3)."""

from textual.screen import Screen
from textual.containers import Vertical
from textual.binding import Binding
from textual.widgets import Footer

from pieces_os_client.wrapper import PiecesClient

from .chat_panel import ChatPanel
from .chat_input import ChatInput, MessageSubmitted


class SimpleChatView(Screen):
    """Simple chat view for testing our components."""

    BINDINGS = [
        Binding("ctrl+q", "quit", "Quit"),
    ]

    DEFAULT_CSS = """
    SimpleChatView {
        layout: vertical;
    }

    SimpleChatView Vertical {
        width: 100%;
        height: 100%;
    }
    """

    def __init__(self, pieces_client: PiecesClient, **kwargs):
        super().__init__(**kwargs)
        self.pieces_client = pieces_client
        self.chat_panel = None
        self.chat_input = None

    def compose(self):
        """Compose the view."""
        with Vertical():
                self.chat_panel = ChatPanel()
                yield self.chat_panel

                self.chat_input = ChatInput()
                yield self.chat_input

        yield Footer()

    def on_mount(self) -> None:
        """Initialize the view."""
        # Show welcome message
        self.chat_panel.show_welcome()

    def on_message_submitted(self, message: MessageSubmitted) -> None:
        """Handle user message submission."""
        # Add user message
        self.chat_panel.add_message("user", message.text)

        # Show thinking indicator
        self.chat_panel.add_thinking_indicator()

        # In Part 3, we'll connect this to the streaming handler
        # For now, just simulate a response
        self.simulate_response(message.text)

    def simulate_response(self, user_message: str):
        """Simulate a response (will be replaced with real streaming in Part 3)."""
        import asyncio

        async def add_response():
            await asyncio.sleep(1)
            self.chat_panel.add_message(
                "assistant",
                f"Echo: {user_message}\n\n(Real AI responses coming in Part 3!)"
            )

        asyncio.create_task(add_response())

💡 Note: This is a simplified version for testing. In Part 3, we'll replace simulate_response() with real streaming from Pieces OS and add the conversations list sidebar.

Step 5: Create the Main Application

Now let's create src/pieces_copilot_tui/app.py:

# src/pieces_copilot_tui/app.py
"""Pieces Copilot TUI - A Terminal User Interface for Pieces OS."""

from textual.app import App

from pieces_os_client.wrapper import PiecesClient
from .chat_view import SimpleChatView


class PiecesCopilotTUI(App):
    """Pieces Copilot TUI (Basic version for Part 2)."""

    # CSS styles for the entire app
    DEFAULT_CSS = """
    Screen {
        background: $background;
        color: $text;
    }

    .error {
        color: $error;
        text-style: bold;
    }

    .success {
        color: $success;
        text-style: bold;
    }
    """

    def __init__(self, pieces_client: PiecesClient = None, **kwargs):
        super().__init__(**kwargs)
        self.pieces_client = pieces_client or PiecesClient()
        self.chat_view = None

    def on_mount(self) -> None:
        """Initialize the application."""
        self.title = "Pieces Copilot TUI"

        # Initialize Pieces client
        if not self.pieces_client.connect_websocket():
            self.notify("Failed to connect to Pieces OS", severity="error")
            self.exit()
            return

        # Create and push the simple chat view
        self.chat_view = SimpleChatView(pieces_client=self.pieces_client)
        self.push_screen(self.chat_view)


def run_tui():
    """Run the Pieces Copilot TUI application."""
    # Initialize Pieces client
    pieces_client = PiecesClient()

    # Check if Pieces OS is running
    if not pieces_client.is_pieces_running():
        print("Error: Pieces OS is not running. Please start Pieces OS first.")
        return

    app = PiecesCopilotTUI(pieces_client=pieces_client)
    app.run()


if __name__ == "__main__":
    run_tui()

Key points:

We inject PiecesClient for better testability
CSS is used for styling (similar to web CSS!)
on_mount() is called when the app starts
Error handling for Pieces OS connection

Step 6: Create the Package Initialization

Create src/pieces_copilot_tui/__init__.py to make it a proper Python package:

# src/pieces_copilot_tui/__init__.py
"""Pieces Copilot TUI Package."""

__version__ = "0.1.0"

from .app import PiecesCopilotTUI, run_tui

__all__ = ["PiecesCopilotTUI", "run_tui"]

This allows the relative imports (.chat_message, .chat_panel, etc.) to work correctly and exports the main functions.

Step 7: Create the Module Entry Point

Create src/pieces_copilot_tui/__main__.py to make the package executable:

# src/pieces_copilot_tui/__main__.py
"""
Command-line entry point for Pieces Copilot TUI.

This allows the package to be run with: python -m pieces_copilot_tui
"""

from .app import run_tui

if __name__ == "__main__":
    run_tui()

This is the Pythonic way to make a package runnable with python -m package_name.

Step 8: Run Your Basic TUI

The cleanest way to run the TUI is from the src directory:

# Navigate to src directory
cd src

# Run the module
python -m pieces_copilot_tui

That's it! No complex setup, no PYTHONPATH manipulation, just clean Python!

What you should see:

A welcome message in the center
An input box at the bottom
You can type messages and see them displayed
Simulated responses after 1 second
Press Ctrl+Q to quit

Recap

In Part 2, we built the foundation of our TUI:

Files Created

✅ src/pieces_copilot_tui/chat_message.py - Individual message widget
✅ src/pieces_copilot_tui/chat_panel.py - Scrollable message display
✅ src/pieces_copilot_tui/chat_input.py - User input widget
✅ src/pieces_copilot_tui/chat_view.py - Simple view (basic version)
✅ src/pieces_copilot_tui/app.py - Main application
✅ src/pieces_copilot_tui/__init__.py - Package definition
✅ src/pieces_copilot_tui/__main__.py - Module entry point

What's Next in Part 3?

In Part 3: Advanced Features & Integration, we'll add:

📋 Conversations List - Sidebar showing all your chats
🌊 Real Streaming - Connect to Pieces OS for actual AI responses
🎯 Full Orchestration - Complete chat_view.py with all features
🧠 LTM Support - Long-Term Memory toggle
⚡ Advanced Features - Delete, rename, create conversations
🧪 Testing & Optimization - Make it production-ready

Preview of Part 3 components:

chats_list.py - Full conversations management
streaming_handler.py - Real-time streaming from Pieces
chat_view.py - Complete orchestrator (expanded version)
Testing, debugging, and performance tips

Troubleshooting

Textual Not Found

pip install textual>=0.47.0

UI Not Updating

# Make sure you're using self.mount() to add widgets
self.mount(message)

# Not just appending to a list

Ready for Part 3?

You now have a working TUI foundation! In Part 3, we'll transform this into a fully-featured Pieces Copilot interface with real AI streaming, conversations management, and all the bells and whistles.

Continue to Part 3 when you're ready! 🚀

Building a TUI with Pieces SDK

Pieces 🌟 — Mon, 02 Feb 2026 17:57:03 +0000

Building a Pieces Copilot TUI - Part 1: Getting Started with PiecesOS SDK

Note: This tutorial is part of the Pieces CLI. We welcome contributions! Feel free to open issues, submit PRs, or suggest improvements.

Introduction

In this two-part tutorial series, we'll build a fully functional Terminal User Interface (TUI) for Pieces Copilot from scratch. In Part 1, we'll explore the PiecesOS SDK and learn how to interact with PiecesOS through coding. In Part 2, we'll create a beautiful TUI using Textual.

What we'll build:

A chat interface with streaming responses
Chat management (create, view, delete)
Long-Term Memory (LTM) support
Real-time UI updates

Prerequisites:

Python 3.8+ (check with python --version)
PiecesOS installed and running (Download here)

Step 1: Setting Up Your Environment

Open or create a new folder for this project. You can name it whatever you’d like. Inside of the folder, follow the steps below:

Create a Virtual Environment

First, let's create an isolated Python environment for our project. Open up a command terminal, and enter the appropriate commands one by one below:


# Create a virtual environment
python -m venv venv

# Activate the virtual environment
## On macOS/Linux:
source venv/bin/activate

## On Windows:
venv\Scripts\activate

Install Dependencies

Still inside of your project folder, create a file called requirements.txt:

# PiecesOS Python SDK for interacting with PiecesOS
pieces-os-client>=3.0.0

# Textual TUI framework for building terminal user interfaces
textual[syntax]>=5.3.0

Inside of a command terminal, install the dependencies:

pip install -r requirements.txt

Step 2: Connecting to PiecesOS

Initialize the Pieces Client

The PiecesClient is your gateway to PiecesOS. Let's create a simple script to connect. Create a new file called test_connection.py:

# test_connection.py
from pieces_os_client.wrapper import PiecesClient

# Initialize the client
client = PiecesClient()

# Check if PiecesOS is running
if client.is_pieces_running():
    print("✅ Connected to PiecesOS!")
    print(f"Version: {client.version}")
else:
    print("❌ PiecesOS is not running. Please start it first.")

Go ahead and test it:

python test_connection.py

What's happening here?

PiecesClient() automatically discovers your PiecesOS instance port
is_pieces_running() checks if PiecesOS is accessible
The client handles port scanning and WebSocket connections

Step 3: Working with Chats

Now, let’s start working on creating and getting chats! First we’ll load all of your chats, then create a new chat, and then get that new chat:

List All Chats

# Add this section to the inside of 'if client.is_pieces_running()', as this all should only run if Pieces can connect.

# Get all chats
chats = client.copilot.chats()

print(f"Found {len(chats)} chats:")
for chat in chats:
    print(f"  - {chat.name}: {chat.summary}")

Create a New Chat

# Place this right below the code you just added above

# chats are created automatically when you ask a question
# without setting an active chat

# Clear current chat to create a new one when we call the stream_question method
client.copilot.chat = None

# You can also do this if you want to create one on the spot
# client.copilot.create_chat("My awesome chat")


# Ask a question - this will create a new chat
client.copilot.stream_question("What is Python?")

Load chat Messages

To make our new chat actually print, we have to print each raw_content that comes back from PiecesOS:

# Add this after 'client.copilot.stream_question("What is Python?")'

# Get all messages in the chat
messages = chats[0].messages()

for msg in messages:
    print(f"{msg.role}: {msg.raw_content}")

Step 4: Asking Questions with Streaming

One of the most powerful features is streaming responses. Instead of waiting for the entire response, you get chunks as they're generated, like you’d see in ChatGPT.

Basic Streaming Example

from pieces_os_client.wrapper import PiecesClient

client = PiecesClient()

def handle_stream(response):
    """Callback function for streaming responses."""
    status = response.status

    if status == "INITIALIZED":
        print("🤔 Thinking...")

    elif status == "IN-PROGRESS":
        # Get the text chunks
        if response.question and response.question.answers:
            for answer in response.question.answers.iterable:
                if answer.text:
                    print(answer.text, end='', flush=True)

    elif status == "COMPLETED":
        print("\n✅ Done!")

    elif status == "FAILED":
        print(f"\n❌ Error: {response.error_message}")

# Register the callback
if client.copilot.ask_stream_ws:
    client.copilot.ask_stream_ws.on_message_callback = handle_stream

# Ask a question
client.copilot.stream_question("Explain Python decorators")

⚠️ Error Handling Tip: In production, we will always handle WebSocket disconnections gracefully. The SDK will attempt to reconnect, but you should inform users about connection status.

Understanding the Streaming Flow

The streaming response goes through several states:

INITIALIZED: Copilot is preparing to respond
IN-PROGRESS: Streaming text chunks
COMPLETED: Response is complete
FAILED/STOPPED/CANCELED: Something went wrong

Step 5: Handling Streaming Properly

Let's create a more robust streaming handler:

# streaming_handler.py
from pieces_os_client.wrapper import PiecesClient
from typing import Callable, Optional

class StreamingHandler:
    """Handles streaming responses from Pieces Copilot."""
    def __init__(
        self,
        pieces_client: PiecesClient,
        on_thinking_started: Callable[[], None],
        on_text_chunk: Callable[[str], None],
        on_completed: Callable[[], None],
        on_error: Callable[[str], None],
    ):
        self.pieces_client = pieces_client
        self.on_thinking_started = on_thinking_started
        self.on_text_chunk = on_text_chunk
        self.on_completed = on_completed
        self.on_error = on_error

        self._current_response = ""

        # Register callback
        if self.pieces_client.copilot.ask_stream_ws:
            self.pieces_client.copilot.ask_stream_ws.on_message_callback = (
                self._handle_stream
            )

    def ask_question(self, query: str):
        """Ask a question and handle streaming."""
        self._current_response = ""
        self.on_thinking_started()
        self.pieces_client.copilot.stream_question(query)

    def _handle_stream(self, response):
        """Internal stream handler."""
        status = response.status

        if status == "IN-PROGRESS":
            if response.question and response.question.answers:
                for answer in response.question.answers.iterable:
                    if answer.text:
                        if not self._current_response:
                            # First chunk
                            self._current_response = answer.text
                        else:
                            # Subsequent chunks
                            self._current_response += answer.text

                        self.on_text_chunk(answer.text)

        elif status == "COMPLETED":
            self.on_completed()
            self._current_response = ""

        elif status in ["FAILED", "STOPPED", "CANCELED"]:
            error_msg = getattr(response, "error_message", "Unknown error")
                self.on_error(error_msg)
            self._current_response = ""

        except (AttributeError, ConnectionError, ValueError) as e:
            # Handle specific streaming errors
            self.on_error(str(e))
            self._current_response = ""

💡 Best Practice: Use specific exception types rather than broad except Exception handlers. This makes debugging easier and prevents masking unexpected errors.

Using the StreamingHandler

# Example usage
client = PiecesClient()

def on_thinking():
    print("🤔 Thinking...")

def on_chunk(text):
    print(f"{text}", end='', flush=True)

def on_done():
    print("\n✅ Done!")
    client.close()

def on_error(error):
    print(f"\n❌ Error: {error}")

handler = StreamingHandler(
    pieces_client=client,
    on_thinking_started=on_thinking,
    on_text_chunk=on_chunk,
    on_completed=on_done,
    on_error=on_error,
)

# Ask a question
handler.ask_question("What are Python generators?")

Step 6: Working with Long-Term Memory (LTM)

LTM allows Pieces to remember context across chats, making responses more personalized.

Check LTM Status

# Check if LTM system is running - place after ensure_initialization()
    if client.copilot.ltm.is_enabled():
        print("✅ LTM system is running")
    else:
        print("❌ LTM system is not available")
        # client.copilot.ltm.enable()

    # Check if current chat has LTM enabled - place after LTM system check
    if client.copilot.ltm.is_chat_ltm_enabled:
        print("✅ Chat LTM is enabled")
    else:
        print("❌ Chat LTM is disabled")

Toggle LTM for Chat

 # Enable LTM for current chat - place after LTM checks
 if client.copilot.ltm.is_enabled():
     client.copilot.ltm.enable()
     print("✅ Chat LTM enabled")
 else:
     print("❌ LTM system must be running first")

 # Disable LTM for current chat - place after enable LTM
    client.copilot.chat_disable_ltm()
    print("✅ Chat LTM disabled")

Step 7: Managing chats

Delete a chat

# Get a chat to delete - place after getting all chats
chat_to_delete = chats[0]

# Delete a chat - place after getting chat to delete
print(f"Deleting: {chat_to_delete.name}")
chat_to_delete.delete()

Rename a chat

# Get a chat to rename - place after getting all chats
chat = chats[0]

# Rename a chat - place after getting chat
chat.name = "My New Chat Name"
print(f"✅ Renamed to: {chat.name}")

Common Issues & Troubleshooting

### "Connection refused" Error

✅ Ensure PiecesOS is running
✅ Check if port 39300 is available

### "Module not found" Error

✅ Activate your virtual environment
✅ Reinstall dependencies: pip install -r requirements.txt
✅ Check Python version: python --version (needs 3.8+)

### Streaming Not Working

✅ Ensure callback is registered before stream_question()
✅ Check that PiecesOS is connected and responsive

Recap

In Part 1, we learned:

✅ Setting up a Python virtual environment
✅ Installing the PiecesOS SDK
✅ Connecting to PiecesOS
✅ Working with chats
✅ Handling streaming responses
✅ Managing Long-Term Memory
✅ Creating a robust streaming handler

Next Steps

In Part 2, we'll use everything we learned to build a beautiful Terminal User Interface (TUI) with:

Split-pane layout for chats and messages
Real-time streaming chat interface
Interactive widgets and keyboard shortcuts
Proper state management and UI updates

Useful Resources

Ready for Part 2? Let's build the TUI! 🚀

The Scale Trap: How AI's Biggest Win Became Its Biggest Problem

Pieces 🌟 — Fri, 19 Dec 2025 14:58:03 +0000

What happens when an entire field forgets everything it learned in the rush to chase one breakthrough?

The AI community is experiencing collective amnesia. We're so focused on making language models bigger that we've forgotten the diverse research that got us here in the first place. This isn't just about nostalgia - it's about understanding why our current approach is hitting hard limits, and what we need to remember to move forward.

Let's trace how we got here, what we lost along the way, and where the most interesting work is happening now.

The Golden Age Nobody Remembers

After AlexNet won ImageNet in 2012, AI research exploded in every direction. This wasn't just about making networks deeper - it was a multi-front advance across fundamentally different approaches to intelligence.

The diversity was staggering:

NLP foundations: Word2Vec gave us semantic embeddings, LSTMs handled sequential data
Generative models: GANs and VAEs competed with totally different philosophies
Strategic AI: Deep RL conquered Atari, Go (AlphaGo), and StarCraft II
Learning efficiency: Meta-learning (MAML) and self-supervised learning tackled data scarcity
Scientific inquiry: XAI, Bayesian methods, adversarial attacks revealed model limitations

This was AI's Cambrian Explosion - tons of different species competing, each solving problems in their own way. Then everything changed.

The Bet That Changed Everything

In 2017, "Attention Is All You Need" introduced the Transformer. The architecture itself was clever, but OpenAI saw something bigger: an engine built for industrial-scale computation.

Their hypothesis was radical: scale alone could trigger a phase transition from pattern matching to genuine reasoning.

The progression was methodical:

The GPT Evolution

GPT-1 established the recipe: pre-training + fine-tuning
GPT-2 showed multitask learning emerging from scale
GPT-3 (175B parameters) demonstrated in-context learning that felt like a paradigm shift

But the real earthquake was ChatGPT, followed by GPT-4 in 2023. This wasn't a research demo anymore - it was a genuinely useful assistant. The bet had paid off spectacularly.

How Success Killed Diversity

GPT-4's success created a gravitational collapse. The entire field got pulled into a single race down the scaling highway. This is where the amnesia began.

Within 2-3 years, researchers could build entire careers in LLM research without deep knowledge of alternative architectures or learning frameworks.

The "Scaling Laws" paper codified this into engineering: invest X compute, get predictable Y improvement. Innovation shifted from algorithmic creativity to capital accumulation.

The Incentive Trap

The monoculture became self-reinforcing:

PhD students: Fastest path to publication is LLM research
Labs: Funding follows the hype
Companies: Existential race for market dominance
Result: Exploring alternative approaches became career suicide

What gets celebrated now? Clever workarounds for LLM limitations:

Prompt Engineering: Crafting inputs for opaque models
RAG: Patching hallucination and knowledge gaps
PEFT (LoRA): Making massive models slightly more adaptable

These are valuable techniques, but they're all downstream fixes. We're accepting the scaled Transformer as gospel instead of questioning the foundation.

The Technical Debt Comes Due

Just as the monoculture peaked, its fundamental limitations became impossible to ignore. More scale can't solve these problems.

Problem 1: The Quadratic Wall

The Transformer's self-attention scales quadratically with sequence length. This creates a hard limit on context windows - analyzing a full codebase, book, or video becomes prohibitively expensive.

The revival: Architectures like Mamba and RWKV achieve linear-time scaling by bringing back recurrent principles. They prove attention isn't all you need.

Problem 2: Running Out of Internet

The scaling hypothesis assumed infinite high-quality data. We're hitting the limits:

Data exhaustion: The supply of quality text is finite
Model collapse: Training on AI-generated content degrades performance

The counter-move: Microsoft's Phi series flips the script. By training smaller models on curated, "textbook-quality" data, they match models 25x their size. Quality beats quantity.

Problem 3: Centralization

A few labs control the frontier. This sparked a grassroots response: the Local AI movement.

Enabled by open models (Meta's Llama) and efficient inference (VLLM), developers are running powerful models on consumer hardware. This creates evolutionary pressure for efficiency - models need to be small and fast, not just powerful.

The Path Forward

The scale era unlocked real capabilities. LLMs are genuinely useful tools. But the amnesia it created - the narrowing of our field's intellectual horizons - is holding us back.

The most interesting work now is happening at the intersection of old and new:

Architectural diversity: Linear-time alternatives to attention
Data science: Quality curation over quantity scraping
Efficiency research: Models that run locally, not just in datacenters
Hybrid approaches: Combining LLMs with symbolic reasoning, retrieval, and other paradigms

We're not abandoning the lessons of scale. We're rediscovering that the forgotten paths - architectural diversity, data-centric training, algorithmic efficiency - are essential for the next phase.

The future won't be a simple extrapolation of scaling laws. It'll be a new synthesis: the raw power discovered through scale, combined with the diversity and ingenuity that defined AI's golden age.

What's your take? Are you working on alternatives to the scaling paradigm? Have you hit these limitations in production? Drop your experiences in the comments.

Tags: #ai #machinelearning #llm #architecture

Building a Daily Productivity App with Pieces - Part 2: Adding AI Intelligence with Gemini

Pieces 🌟 — Wed, 17 Dec 2025 15:39:12 +0000

Welcome back! In Part 1, we built a complete PiecesOS service that:

Connects to PiecesOS and maintains a WebSocket connection
Fetches and caches workstream summaries grouped by day
Extracts the summary content from annotations

Now we have all of the infrastructure in place. But here's the thing: having raw summaries with their content is cool, but it's not exactly... useful. I mean, I have all this text about what I did, but what did I actually accomplish today?

That's where Part 2 comes in. We're going to use Google's Gemini AI to transform those raw summaries into actual insights. Think:

What did I work on?
Which projects did I touch?
Who did I collaborate with?
What should I remember for tomorrow?

The Challenge

Thanks to Part 1, we now have access to rich summary data using SummaryWithContent:

// lib/models/daily_recap_models.dart

// Don't forget the imports!
import 'dart:convert';
import 'package:google_generative_ai/google_generative_ai.dart';
import '../models/daily_recap_models.dart';

SummaryWithContent {
  id: "4f302bfd-f3c2-4f85-aa79-e7cb314e111d",
  title: "Implemented WebSocket sync",
  content: "# Project XYZ\nFixed critical authentication bug in OAuth token refresh.
            Implemented real-time WebSocket synchronization with automatic 
            reconnection. Pair programmed with Bob on the WebSocket integration...",
  timestamp: 2025-11-04 14:32:15
}

This is great! But it's still just raw text. What we want is structured and actionable insights:

SUMMARY:
   Successfully fixed critical authentication bug and implemented 
   real-time WebSocket synchronization for better data flow.

PROJECTS:
   ✅ Authentication Service [completed]
      Fixed OAuth token refresh logic

   🔄 WebSocket Integration [in_progress]
      Implemented real-time sync with automatic reconnection

PEOPLE WORKED WITH:
   • Alice (code review)
   • Bob (pair programming)

REMINDERS:
   ⚠️  Test WebSocket with production load
   ⚠️  Update documentation for new auth flow
NOTES:
  - Send a message in Google Chat about the progress!

See the difference? One is data, the other is information.

Enter Gemini

Google's Gemini API is perfect for this. It can:

Understand natural language
Extract structured information
Return JSON (which is exactly what we need!)
Process multiple summaries at once

Setting Up

First, add the Gemini SDK to pubspec.yaml below the ‘git:’dependency:

dependencies:
  google_generative_ai: ^0.4.6

You'll also need an API key. Get one from Google AI Studio – it's free for reasonable usage!

Building the Daily Recap Service

Let's create a new service: lib/services/daily_recap_service.dart.

class DailyRecapService {
  final GenerativeModel _model;

  DailyRecapService({required String apiKey})
      : _model = GenerativeModel(
          model: 'gemini-2.5-flash-lite',  // Fast, efficient, and cost-effective!
          apiKey: apiKey,
          generationConfig: GenerationConfig(
            temperature: 0.7,  // Balanced creativity
            topK: 40,
            topP: 0.95,
            maxOutputTokens: 2048,
            responseMimeType: 'application/json',
          ),
        );

  Future<DailyRecapData> generateDailyRecap({
    required DateTime date,
    required List<SummaryWithContent> summaries,
  }) async {
    // We'll build this step by step!
  }
}

-** gemini-2.5-flash-lite:** faster and more cost-effective
-** temperature: 0.7: **Not too creative, not too rigid

Now, let's build the generateDailyRecap function piece by piece.

The Prompt Engineering

Crafting the right prompt is an art. Here are some important tips:

Avoid Vague prompts

Analyze these summaries and tell me what I did today.

Result: Always try to be specific. AI does not read your mind… yet! (Pieces does read your mind, but whatever 😉)

Better Structure

Return JSON with: summary, projects, people.

Result: Always say what do you expect the AI to return to be able to correctly parse it

Show Some Examples

Extract and organize into this EXACT JSON format:
{
  "summary": "Brief 1-2 sentence overview",
  "people": ["Person1", "Person2"],
  "projects": [
    {
      "name": "Project Name",
      "description": "What was done",
      "status": "in_progress"  // or "completed" or "not_started"
    }
  ],
  "reminders": ["Reminder 1"],
  "notes": ["Important insight"]
}

Result: AI is similar to human best way to understand by examples

Our Beautiful Prompt

Here's the final prompt we’ll use:

// lib/services/daily_recap_service.dart

  // Add below DailyRecapService() and above Future<>


  /// Build the prompt for Gemini
  String _buildPrompt(DateTime date, String context) {
    final dateStr =
        '${date.year}-${date.month.toString().padLeft(2, '0')}-${date.day.toString().padLeft(2, '0')}';

    return '''You are an AI assistant analyzing a developer's workstream summaries for the day: $dateStr.

Based on the following workstream summaries, extract and organize the information into a structured daily recap.

WORKSTREAM SUMMARIES:
$context

Your task is to analyze these summaries and create a comprehensive daily recap with the following information:

1. **summary** (string, 1-2 sentences): A brief overview of what was accomplished today. Focus on the main achievements and work done.

2. **people** (array of strings): List of people mentioned or collaborated with. Look for names, @mentions, or collaboration indicators. Can be empty if no one is mentioned.

3. **projects** (array of objects): Projects worked on today. Each project should have:
   - **name** (string): Project or feature name
   - **description** (string): Brief description of what was done
   - **status** (string): One of: "completed", "in_progress", or "not_started"

4. **reminders** (array of strings): Action items, TODOs, or things to remember for later. Look for phrases like "need to", "should", "TODO", "remember to", etc. Can be empty.

5. **notes** (array of strings): Important observations, learnings, or technical notes from the day. Look for insights, discoveries, or important information. Can be empty.

IMPORTANT GUIDELINES:
- Be concise but informative
- Extract actual information from the summaries, don't make things up
- If a category has no relevant information, use an empty array [] or empty string ""
- For project status: use "completed" if the work is done, "in_progress" if actively working on it, "not_started" if mentioned but not begun
- People names should be just the name (e.g., "Alice", "Bob")
- Keep descriptions clear and specific

Return ONLY valid JSON in this exact format:
{
  "summary": "Brief 1-2 sentence overview",
  "people": ["Person1", "Person2"],
  "projects": [
    {
      "name": "Project Name",
      "description": "What was done",
      "status": "in_progress"
    }
  ],
  "reminders": ["Reminder 1", "Reminder 2"],
  "notes": ["Note 1", "Note 2"]
}
''';
  }

Why this works:

1.Clear role: "You are an AI assistant..."
2.Specific format: Exact JSON structure
3*. Examples:* Shows what we want
4.** Constraints:** "Don't make things up", "Empty arrays if no data"
5.** Enum values:** Explicit status options

These two helper methods,_buildPrompt and _buildSummariesContext(we'll see next), are what power our analysis. But where do they fit in the actual application?

Sending Rich Context to Gemini

Now that we have the actual summary content, we can build a rich prompt:

 // lib/services/daily_recap_service.dart

 // Add below _buildPrompt() and above generateDailyRecap()

 String _buildSummariesContext(List<SummaryWithContent> summaries) {
    final buffer = StringBuffer();

    for (int i = 0; i < summaries.length; i++) {
      final summary = summaries[i];
      buffer.writeln('Summary ${i + 1}:');
      buffer.writeln('  ID: ${summary.id}');
      buffer.writeln('  Title: ${summary.title}');
      buffer.writeln(
        '  Time: ${summary.timestamp.hour.toString().padLeft(2, '0')}:${summary.timestamp.minute.toString().padLeft(2, '0')}',
      );
      buffer.writeln('  Content: ${summary.content}');
      buffer.writeln();
    }

    return buffer.toString();
  }

This formats each summary with structured labels and metadata, giving Gemini way more context to work with

Let’s add an empty factory method to create an empty DailyRecapData

// lib/services/daily_recap_service.dart

// Add below _buildPrompt() and above generateDailyRecap() 
 factory DailyRecapData.empty(DateTime date) {
    return DailyRecapData(
      date: date,
      summary: '',
      people: [],
      projects: [],
      reminders: [],
      notes: [],
    );
  }

Handling the Response

Gemini returns JSON (because we set responseMimeType), so parsing is straightforward:

// lib/services/daily_recap_service.dart

// Replace the comment in generateDailyRecap() with this:

try {
  final response = await _model.generateContent([Content.text(prompt)]);
  print("Gemini response received. ${response.text}");  // Debug output
  final rawText = response.text ?? '{}';

  // Extract JSON from markdown code blocks
  final jsonText = _extractJsonFromMarkdown(rawText);
  final data = jsonDecode(jsonText) as Map<String, dynamic>;

  return DailyRecapData.fromJson(date, data); // ... generate recap
} catch (e) {
  print('Error generating recap: $e');
  return DailyRecapData.empty(date);  // Safe fallback
}

Putting It All Together: The Complete generateDailyRecap Function

Now that we've seen all the pieces, here's how they fit together in the actual generateDailyRecap function (how yours should look 😉):

Future<DailyRecapData> generateDailyRecap({
  required DateTime date,
  required List<SummaryWithContent> summaries,
}) async {
  // Step 1: Handle edge case - no summaries
  if (summaries.isEmpty) {
    return DailyRecapData.empty(date);
  }

  // Step 2: Build context from summaries using our helper method
  final context = _buildSummariesContext(summaries);

  // Step 3: Craft the prompt using our prompt builder
  final prompt = _buildPrompt(date, context);

  try {
    // Step 4: Send to Gemini and get response
    final response = await _model.generateContent([Content.text(prompt)]);
    print("Gemini response received. ${response.text}");

    // Step 5: Parse the JSON response
    final jsonText = response.text ?? '{}';
    final data = jsonDecode(jsonText) as Map<String, dynamic>;

    // Step 6: Convert to our data model and return
    return DailyRecapData.fromJson(date, data);
  } catch (e) {
    print('Error generating daily recap: $e');
    rethrow; // Let the UI handle the error
  }
}

See how it flows?

Check for empty summaries
Build the context string from all summaries
Create the prompt with instructions
Send to Gemini
Parse the JSON response
Return structured data (or throw error)

The Data Models

I created clean data classes to work with:

// lib/models/daily_recap_models.dart

// Add before SummaryWithContent class and after ProjectStatus {}

class ProjectData {
  final String name;
  final String description;
  final ProjectStatus status;  // enum: completed, inProgress, notStarted

  ProjectData({
    required this.name,
    required this.description,
    required this.status,
  });

  factory ProjectData.fromJson(Map<String, dynamic> json) {
    return ProjectData(
      name: json['name'] as String,
      description: json['description'] as String,
      status: _statusFromString(json['status'] as String),
    );
  }

  static ProjectStatus _statusFromString(String status) {
    switch (status) {
      case 'completed':
        return ProjectStatus.completed;
      case 'in_progress':
        return ProjectStatus.inProgress;
      case 'not_started':
        return ProjectStatus.notStarted;
      default:
        return ProjectStatus.notStarted;
    }
  }
}

class DailyRecapData {
  final DateTime date;
  final String summary;
  final List<String> people;
  final List<ProjectData> projects;
  final List<String> reminders;
  final List<String> notes;

  DailyRecapData({
    required this.date,
    required this.summary,
    required this.people,
    required this.projects,
    required this.reminders,
    required this.notes,
  });

  factory DailyRecapData.fromJson(DateTime date, Map<String, dynamic> json) {
    return DailyRecapData(
      date: date,
      summary: json['summary'] as String? ?? '',
      people: (json['people'] as List<dynamic>?)
              ?.map((e) => e as String)
              .toList() ??
          [],
      projects: (json['projects'] as List<dynamic>?)
              ?.map((e) => ProjectData.fromJson(e as Map<String, dynamic>))
              .toList() ??
          [],
      reminders: (json['reminders'] as List<dynamic>?)
              ?.map((e) => e as String)
              .toList() ??
          [],
      notes: (json['notes'] as List<dynamic>?)
              ?.map((e) => e as String)
              .toList() ??
          [],
    );
  }
}

This gives us type safety and makes it easy to work with the data later.

Look at what Gemini did:

✅ Understood that I was working on "Pieces OS Integration"
✅ Correctly identified it as "completed"
✅ Extracted actual reminders from my work
✅ Pulled out technical notes I discovered
✅ Wrote a coherent summary of the day

And it did all this from just timestamps and titles!

Real-World Example: What Gemini Generated

Here's an actual JSON response that Gemini generated from my workstream summaries:

{
  "summary": "Today's work focused on enhancing user experience for video content, developing a tag generator, testing Flutter capabilities, and reviewing code and infrastructure. Significant progress was made on persona generation for AI training data.",
  "people": [],
  "projects": [
    {
      "name": "Video Analytics & User Experience",
      "description": "Analyzed YouTube video analytics for 'Pieces' content and discussed strategies for improving new user experience by leveraging context and memory,
 considering phased UI exposure and temporary access keys.",
      "status": "in_progress"
    },
    {
      "name": "Tag Generator",
      "description": "Generated a Python script for thematic tagging.",
      "status": "completed"
    },
    {
      "name": "Flutter macOS Dynamic Library Loading",
      "description": "Demonstrated Flutter's macOS dynamic library loading and confirmed clipboard monitoring functionality and its integration with long-term memory.",
      "status": "in_progress"
    },
    {
      "name": "AI Persona Generation",
      "description": "Discussed UI for AI persona generation and initiated content compilation for a partner. Presented the 'persona-query-tag-dataset-gen' project, det
ailing the creation of realistic user personas for the Pieces AI assistant, showcasing comprehensive attributes and providing an example of 'Anja Vestergaard'.",
      "status": "in_progress"
    },
    {
      "name": "ML Training & Django API",
      "description": "Addressed an `UnboundLocalError` in ML training and validated nested task management for a Django API.",
      "status": "completed"
    },
    {
      "name": "Infrastructure Upgrades & Containerization",
      "description": "Reviewed infrastructure upgrades, containerization strategies, cost optimizations, and bug fixes across multiple services, including timezone and 
user invitation flows.",
      "status": "in_progress"
    }
  ],
  "reminders": [],
  "notes": [
    "Leveraging context and memory for new user experience improvements.",
    "Clipboard monitoring functionality confirmed and integrated with long-term memory.",
    "Objective for persona generation project is to generate authentic training data for the AI assistant."
  ]
}

Complete Reference Implementation

For your reference, here's the complete lib/services/daily_recap_service.dart file with everything we've built:

The service initialization with Gemini configuration
The generateDailyRecap function that orchestrates everything
The _buildSummariesContext helper for formatting summaries
The _buildPrompt helper for crafting the AI prompt

Continue reading here.

Building Daily Stand-Up Generator using Pieces API - Part 1: The SDK overview

Pieces 🌟 — Wed, 17 Dec 2025 15:22:08 +0000

- written by Bishoy Hany

The Monday Morning Problem

Imagine the following scenario: Monday, 9 AM. You grab your coffee, join the stand-up call, and suddenly realise: you have absolutely no idea what you did last week.
The weekend wiped your mental cache clean. Was it the React component refactor? Or was that two weeks ago? You vaguely remember fighting with TypeScript errors on Thursday, but what was the actual solution? Your Git commits say 'fix: update logic'—thanks, past you, very helpful.

Here's the thing: your brain isn't built to be a perfect activity log. But your computer? It remembers everything. That's where PiecesOS comes in.

The Goal

Create an app that tracks what work is done every day.

For Part 1, we will work on:

Connecting to PiecesOS
Getting our workstream summaries
Grouping the summaries by day
Keeping everything updated in real-time

We’ll hit a UI later on. Let’s take this one step at a time!

Prerequisites

Dart SDK version 3.8.1 or higher (flutter.dev)
Gemini API key (will mention that in part 2

Setting Up

Let’s begin by creating a new flutter app

flutter create daily_recap_app

Before you can start using the Pieces SDK, you’ll need to add it to your Dart or Flutter project. Open your project’s pubspec.yaml file and add the following under ‘dependencies:’:

dependencies:
  # The Pieces SDK
  pieces_os_client:
    git:
      url: https://github.com/pieces-app/pieces-os-client-sdk-for-dart.git

  web_socket_channel: ^3.0.1

The SDK is automatically generated from our API setup, so it’s mostly made up of code that helps your app talk to Pieces’ api.
We also added the web_socket_channel dependency to connect to the Pieces web-socket

Step 1: Connecting to PiecesOS

Alright, first challenge – how do we even talk to PiecesOS?

Let's first create a new file lib/services/pieces_os_service.dart

import 'dart:async';
import 'dart:convert';
import 'dart:io';
import 'package:pieces_os_client/api.dart';
import 'package:web_socket_channel/web_socket_channel.dart';

class PiecesOSService {
  // Super important Pieces sometimes could run on other ports so you should implement your own port-finding algo
  // here is example using the python sdk
  // https://github.com/pieces-app/pieces-os-client-sdk-for-python/blob/db1fa83f21407323b98d21507a59d1e4f176d064/src/pieces_os_client/wrapper/client.py#L139-L168
  // We gonna use 39300 for simplicity and PiecesOS will be running on port 39300 for most of the time as well
  static const String baseUrl = 'http://localhost:39300';
  static const String websocketUrl = 'ws://localhost:39300';

  // Used to register application
  late final ConnectorApi _connectorApi;
  // Used to retrieve the workstream summaries
  late final WorkstreamSummaryApi _workstreamSummaryApi;

  // Used to retrieve the summary content
  late final AnnotationApi _annotationApi;

  // Cache for workstream summaries grouped by day
  final Map<DateTime, List<WorkstreamSummary>> _summariesByDay = {};

  PiecesOSService() {
    final client = ApiClient(basePath: baseUrl);
    _connectorApi = ConnectorApi(client);
    _workstreamSummaryApi = WorkstreamSummaryApi(client);
    _annotationApi = AnnotationApi(client);
  }
}

Nothing fancy – just setting up our API clients.

Now the fun part – actually connecting:

 // existing code from above
  Future<Application> connectApplication() async {
    final seededApp = SeededTrackedApplication(
      name: ApplicationNameEnum.OPEN_SOURCE,  // Hey Pieces, I'm open source!
      platform: Platform.operatingSystem == "macos"
          ? PlatformEnum.MACOS
          : PlatformEnum.WINDOWS,
      version: "0.0.1",
    );

    final connection = SeededConnectorConnection(
      application: seededApp,
    );

    _context = await _connectorApi.connect(
      seededConnectorConnection: connection,
    );

    print('Successfully connected to Pieces OS!');
    return _context!.application;
  }

Basically, we're saying, "Hey Pieces, I'm a new app, let me in!" and it gives us back a context with our application info.

Step 2: The WebSocket

This websocket specifically sends us all of the IDs for the users' workstream summaries in Pieces upon first connection. After, it sends only the IDs of either newly created summaries, summaries that have updated data, or summaries that were deleted.

Here's the setup:

  // existing code from above
  void _startWebSocketListener() {
    final wsUrl = '$websocketUrl/workstream_summaries/stream/identifiers';
    _wsChannel = WebSocketChannel.connect(Uri.parse(wsUrl));

    print('WebSocket connected to $wsUrl');

    _wsSubscription = _wsChannel!.stream.listen(
      (message) async {
        try {
          // Important: WebSocket sends JSON strings, decode them first!
          final streamedIdentifiers = StreamedIdentifiers.fromJson(
            jsonDecode(message),
          );

          // Loop through each identifier we received
          for (final id in streamedIdentifiers!.iterable) {
            final summaryId = id.workstreamSummary?.id;
            if (summaryId != null) {
              await _fetchAndCacheSummary(summaryId);
            }
          }
        } catch (e) {
          print('Error processing message: $e');
        }
      },
      onDone: () {
        print('WebSocket closed, reconnecting...');
        _reconnectWebSocket();
      },
      onError: (error) {
        print('WebSocket error: $error');
        _reconnectWebSocket();
      },
    );
  }

Here's the beauty of this approach: The WebSocket just keeps until their app shuts down or they manually close it, sending us identifiers. When we first connect, it dumps ALL existing identifiers on us (could be from a year ago!) and stays open to send us new ones as they're created in real-time. One connection handles everything!
Step 3: Fetching and Caching Summaries
Okay, so we have identifiers. Now what? We need to actually fetch the summary data and organize it.

Here's what a WorkstreamSummary looks like (the important parts anyway):

// This is a model defined in the sdks don't add any thing to the code yet.

class WorkstreamSummary {
  String id;              // Unique identifier
  String? name;           // Optional name/title
  GroupedTimestamp created;  // When it was created
  GroupedTimestamp? updated; // When it was last updated
  // ... and a bunch of other fields
}

And GroupedTimestamp is basically:

// This is a model defined in the sdks don't add any thing to the code yet.

class GroupedTimestamp {
  DateTime value;  // The actual timestamp
}

So when we fetch a summary, we need to:

Get the created date
Strip the time from the date
Add the summary to that day's list
Sort everything by time

Here's the code:

 // existing code from above
  Future<void> _fetchAndCacheSummary(String identifier) async {
    final summary = await _workstreamSummaryApi
        .workstreamSummariesSpecificWorkstreamSummarySnapshot(identifier);

    if (summary == null) {
      return;
    }
    // Normalize to day (remove time)
    final createdDate = summary.created.value;
    final dayKey = DateTime(
      createdDate.year,
      createdDate.month,
      createdDate.day,
    );

    // Create day entry if it doesn't exist
    _summariesByDay.putIfAbsent(dayKey, () => []);

    // Check for duplicates (avoid adding the same summary twice)
    final existingIndex = _summariesByDay[dayKey]!
        .indexWhere((s) => s.id == summary.id);

    if (existingIndex != -1) {
      // Update existing
      _summariesByDay[dayKey]![existingIndex] = summary;
    } else {
      // Add new
      _summariesByDay[dayKey]!.add(summary);
    }

    // Sort by time (most recent first)
    _summariesByDay[dayKey]!.sort((a, b) {
      return b.created.value.compareTo(a.created.value);
    });

    // Notify anyone listening
    _summariesStreamController.add(Map.unmodifiable(_summariesByDay));

    print('Cached summary ${summary.id} for day $dayKey');
  }

The _summariesByDay is just aMap<DateTime, List<WorkstreamSummary>> where:

Key: A DateTime set to midnight (e.g., Nov 4, 2025 at 00:00:00)
Value: List of all summaries for that day, sorted by time

Let's also add some useful methods to retrieve a summary

 // existing code
  /// Get summaries for a specific day from cache
  List<WorkstreamSummary> getSummariesForDay(DateTime date) {
    final dayKey = DateTime(date.year, date.month, date.day);
    return _summariesByDay[dayKey] ?? [];
  }

  /// Get all days that have summaries (sorted most recent first)
  List<DateTime> getDaysWithSummaries() {
    final days = _summariesByDay.keys.toList();
    days.sort((a, b) => b.compareTo(a)); // Most recent first
    return days;
  }

Step 4: Keeping Everything in Sync
Speaking of streams – let's add a broadcast stream that others can listen to:

// existing code
final StreamController<Map<DateTime, List<WorkstreamSummary>>>
    _summariesStreamController = StreamController.broadcast();

Stream<Map<DateTime, List<WorkstreamSummary>>> get summariesStream =>
    _summariesStreamController.stream;

So in the UI (when we build it), we can just do:

service.summariesStream.listen((summaries) {
  print('Got new data!');
  // Update UI here
});

Step 4.5: Waiting for Initial WebSocket Sync

Here's a problem: when the app first starts, the WebSocket dumps all existing summaries on us. This could be hundreds of identifiers! If we try to show the UI or generate a recap before they're all loaded, we'll have incomplete data.

But we only need to wait once – on the first message. After that, the WebSocket just sends new updates in real-time and we don't want to block.

Here's the solution using a Completer:

Step 1: Add Fields and Wait Method

Place this code block right after the PiecesOSService() constructor and before the connectApplication() method:

// Add these fields to the class
Completer<void>? _initialSyncCompleter;
bool get _isInitialSyncComplete =>
      _initialSyncCompleter?.isCompleted ?? false;
/// Wait for the initial WebSocket sync to complete
/// Only blocks on the first call, returns immediately after
Future<void> waitForInitialSync() async {
  // If already synced, return immediately
  if (_isInitialSyncComplete) {
    return;
  }

  // If sync is in progress, wait for it
  if (_initialSyncCompleter != null) {
    return _initialSyncCompleter!.future;
  }

  // Start waiting for first sync
  _initialSyncCompleter = Completer<void>();

  return _initialSyncCompleter!.future;
}

Step 2: Update WebSocket Listener

Replace _wsSubscription with this new section:

_wsSubscription = _wsChannel!.stream.listen(
  (message) async {
    try {
      final streamedIdentifiers = StreamedIdentifiers.fromJson(
        jsonDecode(message),
      );

      // Process all identifiers
      for (final id in streamedIdentifiers!.iterable) {
        final summaryId = id.workstreamSummary?.id;
        if (summaryId != null) {
          await _fetchAndCacheSummary(summaryId);
        }
      }

      // Mark initial sync as complete after first message
      if (!_isInitialSyncComplete) {
        _isInitialSyncComplete = true;
        _initialSyncCompleter?.complete();
        print('Initial WebSocket sync complete!');
      }
    } catch (e) {
      print('Error processing message: $e');
    }
  },
  // ... onDone, onError handlers
);

Step 3: Use in App Initialization

Place this code in your app's entry point (typically lib/main.dart, create the file if you don't have it):

import 'services/pieces_os_service.dart';

Future<void> main() async {
  final service = PiecesOSService();

  await service.connectApplication();
  service.startWebSocketListener();

  // Wait for all existing summaries to load
  await service.waitForInitialSync();

  // Now we can safely generate recaps or show UI!
  print('Ready to go! All summaries loaded.');
}

Step 5: Accessing the Actual Summary Content

WorkstreamSummary objects from the SDK don't directly expose the summary text. The actual content is stored in annotations.

Each summary has annotations, and we need to:

Loop through the summary's annotations
Find the one with type SUMMARY
Fetch that annotation using AnnotationApi
Extract the text content

Let's add a method to fetch annotation content:

// At the bottom of pieces_os_service.dart

/// Get the summary content from a workstream summary's annotations
Future<String?> getSummaryContent(WorkstreamSummary summary) async {
  try {
    // Loop through annotations to find the DESCRIPTION type
    for (final annotationRef in summary.annotations?.indices.keys.toList() ?? []) {
      // Fetch the full annotation using AnnotationApi
      final annotation = await _annotationApi
          .annotationSpecificAnnotationSnapshot(annotationRef);

      if (annotation == null) {
        continue;
      }
      // Check if this is a DESCRIPTION type annotation
      if (annotation.type == AnnotationTypeEnum.SUMMARY) {
        // Return the text content - this is the actual summary!
        return annotation.text;
      }
    }

    return null;
  } catch (e) {
    print('Error fetching annotation content for ${summary.id}: $e');
    return null;
  }
}

The SDK separates metadata (ID, timestamp) from content (stored in annotations).

Step 6: Creating a Proper Data Structure

Let's create a proper class for summaries with their content. Add this to lib/models/daily_recap_models.dart:

/// Data class to hold a summary with its content
class SummaryWithContent {
  final String id;
  final String title;
  final String content;        // The actual summary text!
  final DateTime timestamp;

  SummaryWithContent({
    required this.id,
    required this.title,
    required this.content,
    required this.timestamp,
  });
}

Now, let's add a convenient method to get summaries with their content:

// Don't forget to import the model that we just created
import 'package:daily_recap_app/models/daily_recap_models.dart';

// Below existing code in pieces_os_straervice.dart

/// Get summaries with their content for a specific day
Future<List<SummaryWithContent>> getSummariesWithContentForDay(
  DateTime date,
) async {
  final summaries = getSummariesForDay(date);
  final List<SummaryWithContent> summariesWithContent = [];

  for (final summary in summaries) {
    final content = await getSummaryContent(summary);
    if (content != null && content.isNotEmpty) {
      summariesWithContent.add(
        SummaryWithContent(
          id: summary.id,
          title: summary.name,
          content: content,              // Actual text from annotation!
          timestamp: summary.created.value,
        ),
      );
    }
  }

  return summariesWithContent;
}

Perfect! Now we have rich, typed data ready for AI processing and UI display.

Part 1 Summary:

Connect to PiecesOS
Open WebSocket for real-time updates
Fetch and cache summaries
Group by day
Stream updates to UI
Extract summary content from annotations

What is created so far lib/services/pieces_os_service.dart

import 'dart:async';
import 'dart:convert';
import 'dart:io';
import 'package:pieces_os_client/api.dart';
import 'package:pieces_os_client/api_client.dart';
import 'package:web_socket_channel/web_socket_channel.dart';
import 'package:daily_recap_app/models/daily_recap_models.dart';

/// Service to interact with Pieces OS for LTM (Long Term Memory) tracking
class PiecesOSService {
  static const String baseUrl = 'http://localhost:39300';
  static const String websocketUrl = 'ws://localhost:39300';

  // Used to register application
  late final ConnectorApi _connectorApi;
  // Used to retrieve the workstream summaries
  late final WorkstreamSummaryApi _workstreamSummaryApi;

  // Used to retrieve the summary content
  late final AnnotationApi _annotationApi;

  final StreamController<Map<DateTime, List<WorkstreamSummary>>>
      _summariesStreamController = StreamController.broadcast();

  Stream<Map<DateTime, List<WorkstreamSummary>>> get summariesStream =>
      _summariesStreamController.stream;

  // Track initial WebSocket sync
  bool _isInitialSyncComplete = false;
  Completer<void>? _initialSyncCompleter;

  PiecesOSService() {
    final client = ApiClient(basePath: baseUrl);
    _connectorApi = ConnectorApi(client);
    _workstreamSummaryApi = WorkstreamSummaryApi(client);
    _annotationApi = AnnotationApi(client);
  }

  Future<Application> connectApplication() async {
    final seededApp = SeededTrackedApplication(
      name: ApplicationNameEnum.OPEN_SOURCE,  // Hey Pieces, I'm open source!
      platform: Platform.operatingSystem == "macos"
          ? PlatformEnum.MACOS
          : PlatformEnum.WINDOWS,
      version: "0.0.1",
    );

    final connection = SeededConnectorConnection(
      application: seededApp,
    );

    _context = await _connectorApi.connect(
      seededConnectorConnection: connection,
    );

    print('Successfully connected to Pieces OS!');
    return _context!.application;
  }

  /// Wait for the initial WebSocket sync to complete
  /// Only blocks on the first call, returns immediately after
  Future<void> waitForInitialSync() async {
    // If already synced, return immediately
    if (_isInitialSyncComplete) {
      return;
    }

    // If sync is in progress, wait for it
    if (_initialSyncCompleter != null) {
      return _initialSyncCompleter!.future;
    }

    // Start waiting for first sync
    _initialSyncCompleter = Completer<void>();
    return _initialSyncCompleter!.future;
  }

  void _startWebSocketListener() {
    final wsUrl = '$websocketUrl/workstream_summaries/stream/identifiers';
    _wsChannel = WebSocketChannel.connect(Uri.parse(wsUrl));

    print('WebSocket connected to $wsUrl');

    _wsSubscription = _wsChannel!.stream.listen(
      (message) async {
        try {
          // Important: WebSocket sends JSON strings, decode them first!
          final streamedIdentifiers = StreamedIdentifiers.fromJson(
            jsonDecode(message),
          );

          // Loop through each identifier we received
          for (final id in streamedIdentifiers!.iterable) {
            final summaryId = id.workstreamSummary?.id;
            if (summaryId != null) {
              await _fetchAndCacheSummary(summaryId);
            }
          }

          // Mark initial sync as complete after first message
          if (!_isInitialSyncComplete) {
            _isInitialSyncComplete = true;
            _initialSyncCompleter?.complete();
            print('Initial WebSocket sync complete!');
          }
        } catch (e) {
          print('Error processing message: $e');
        }
      },
      onDone: () {
        print('WebSocket closed, reconnecting...');
        _reconnectWebSocket();
      },
      onError: (error) {
        print('WebSocket error: $error');
        _reconnectWebSocket();
      },
    );
  }

  Future<void> _fetchAndCacheSummary(String identifier) async {
    final summary = await _workstreamSummaryApi
        .workstreamSummariesSpecificWorkstreamSummarySnapshot(identifier);

    if (summary == null) {
      return;
    }
    // Normalize to day (remove time)
    final createdDate = summary.created.value;
    final dayKey = DateTime(
      createdDate.year,
      createdDate.month,
      createdDate.day,
    );

    // Create day entry if it doesn't exist
    if (!_summariesByDay.containsKey(dayKey)) {
      _summariesByDay[dayKey] = [];
    }

    // Check for duplicates (avoid adding the same summary twice)
    final existingIndex = _summariesByDay[dayKey]!
        .indexWhere((s) => s.id == summary.id);

    if (existingIndex != -1) {
      // Update existing
      _summariesByDay[dayKey]![existingIndex] = summary;
    } else {
      // Add new
      _summariesByDay[dayKey]!.add(summary);
    }

    // Sort by time (most recent first)
    _summariesByDay[dayKey]!.sort((a, b) {
      return b.created.value.compareTo(a.created.value);
    });

    // Notify anyone listening
    _summariesStreamController.add(Map.unmodifiable(_summariesByDay));

    print('Cached summary ${summary.id} for day $dayKey');
  }

  /// Get summaries for a specific day from cache
  List<WorkstreamSummary> getSummariesForDay(DateTime date) {
    final dayKey = DateTime(date.year, date.month, date.day);
    return _summariesByDay[dayKey] ?? [];
  }

  /// Get all days that have summaries (sorted most recent first)
  List<DateTime> getDaysWithSummaries() {
    final days = _summariesByDay.keys.toList();
    days.sort((a, b) => b.compareTo(a)); // Most recent first
    return days;
  }

  /// Get ries with their content for a specific day
  Future<List<SummaryWithContent>> getSummariesWithContentForDay(
    DateTime date,
  ) async {
    final summaries = getSummariesForDay(date);
    final List<SummaryWithContent> summariesWithContent = [];

    for (final summary in summaries) {
      final content = await getSummaryContent(summary);
      if (content != null && content.isNotEmpty) {
        summariesWithContent.add(
          SummaryWithContent(
            id: summary.id,
            title: summary.name,
            content: content,
            timestamp: summary.created.value,
          ),
        );
      }
    }

    return summariesWithContent;
  }
}

Reference GitHub for viewing the full project.

3 Ways to Actually Improve Your Work Performance (That Changed Everything for Me)

Pieces 🌟 — Wed, 26 Nov 2025 16:11:43 +0000

A year ago, if you'd asked me how to boost productivity, I'd have given you the standard advice: work harder, multitask better, attend more meetings. Turns out, that's exactly the wrong approach.

Real improvement comes from working smarter. Here are three strategies that actually moved the needle for me.

The Context Drift Problem

We're all drowning in information. Emails, meetings, Slack threads, endless to-do lists. The real productivity killer isn't the work itself – it's trying to remember what happened yesterday, why you made that decision, or where that critical feedback lives.

23 minutes – that's how long it takes to refocus after an interruption

This "context drift" destroys more productivity than most people realize. Here's what actually works:

Capture as you go. Don't trust your memory. Write down decisions, key takeaways, and blockers as they happen. Notebook, notes app, shared doc – doesn't matter. Just capture it.

Centralize everything. Scattered notes across five different tools? That's a recipe for losing context. Keep your work information in as few places as possible.

Review regularly. Spend 5-10 minutes at the end of each day scanning your notes. What's moving forward? What's stuck? What needs attention next? This habit surfaces patterns you'd otherwise miss.

Make context visible. Share your context with your team. When everyone understands the "why" behind decisions, collaboration gets smoother and misunderstandings drop.

Automating Context Tracking

Here's the thing – you don't have to do this manually anymore. I use Pieces to automatically track my work across docs, content plans, campaigns, and conversations.

Instead of scrambling to remember what I did last week, I can instantly pull up a timeline of my decisions, blockers, and progress:

This saves me hours every month and gives me confidence that nothing important slips through the cracks. Context tracking goes from manual chore to effortless advantage.

Killing the Reporting Mental Load

Once you've captured your context, the next challenge is reporting on it. Stand-ups, reviews, status updates – they all require sifting through scattered notes and memories. This mental load is one of the most overlooked productivity killers.

Research shows that managing cognitive load is a proven way to improve workplace performance

Here's what works:

Batch your reporting. Don't update constantly. Set specific times for reporting and stick to them.

Keep a running log. Maintain one place where you track accomplishments and blockers throughout the week.

Automate what you can. Use tools that generate reports from your actual work, not from your memory of the work.

Don't aim for perfection. Good enough is good enough for most updates.

The less mental energy you spend on reporting, the more you have for creative problem-solving and actual execution.

How I Automated My Reporting

Before Pieces, I dreaded reporting cycles. Hours spent reconstructing my week, second-guessing what mattered, worrying I'd forget something important.

Now? I just ask Pieces "What did I accomplish this week?" and get a clear, data-driven narrative of my progress:

This automation freed up tons of mental bandwidth. I can focus on higher-impact work instead of administrative overhead.

You can try it yourself – download Pieces, enable Long-Term Memory, and let it capture your workflow for a day or two. You have full control over what's recorded (you can block any apps), and all sensitive data stays on-device.

Once it's set up, try generating your workstream activity. You'll see how quickly you can turn real activity into a clear, ready-to-share update.

Turning Updates Into Growth

With context captured and reporting automated, you can do something powerful: reflect and grow. Use your daily updates not just as a record, but as a mirror to spot patterns, celebrate wins, and identify improvement areas.

Look for patterns. What days are you most productive? What types of tasks drain you? When do blockers typically appear?

Celebrate small wins. Don't wait for big milestones. Acknowledge daily progress to maintain momentum.

Identify growth areas. Where are you consistently stuck? What skills would help you move faster?

Share and seek feedback. Use your updates as conversation starters with your team or manager.

Turning routine updates into actionable insights is the foundation of growth. It's how you move from just "doing the work" to understanding how to improve it.

My Updates Became a Feedback Loop

At first, my daily updates felt like another checkbox. But once I started using Pieces, everything changed. The AI analyzed my updates, surfaced productivity trends, highlighted recurring blockers, and suggested improvement areas:

I discovered I was most productive in the mornings and that meetings were my biggest interruption source. With this data, I adjusted my schedule and advocated for fewer midday meetings.

Result? Noticeable improvements in both output and job satisfaction.

Now every update isn't just a status report – it's a mini case study in how I solve problems, collaborate, and grow. Pieces turns my daily routines into a personalized feedback loop, helping me continuously improve and align my work with long-term goals.

What Actually Works

Improving work performance isn't about longer hours or more meetings. The real breakthroughs came from:

Gaining clarity through context tracking – Making my work visible and accessible
Reducing mental load by automating reporting – Freeing up bandwidth for high-impact work
Turning daily updates into personal insights – Using data to drive continuous improvement

These aren't abstract ideas. They're changes I made through experience, supported by the right tools and honest reflection on my workflow.

If you're looking for ways to improve your work performance, start by making your work visible, your reporting effortless, and your growth intentional. Try Pieces and see the results for yourself.

What strategies have worked for you? Have you found ways to reduce context switching or automate your reporting? Drop your experiences in the comments.

productivity #workperformance #developertools #automation

How I Automated My Standup Meetings (And Why You Should Too)

Pieces 🌟 — Wed, 26 Nov 2025 16:11:08 +0000

Stop wasting 15 minutes every morning on status updates. Here's how I built a system that writes my standups automatically—and actually makes them useful.

If you're like most developers, your morning standup feels like this: scramble to remember what you did yesterday, mumble something about "working on tickets," promise to finish that PR today, then immediately forget what everyone else said.

I used to spend 15 minutes every morning just preparing for a 5-minute standup. Multiply that by 5 days a week, and I was burning over an hour weekly on meeting prep alone.

Here's the thing: standups aren't the problem. The manual status reporting is.

So I automated mine. Completely. And it's been a game-changer.

The Problem: Standups Are Broken (But We Need Them)

Daily standups exist for good reasons:

Keep the team aligned on priorities
Surface blockers before they become disasters
Build accountability and momentum
Create visibility across distributed teams

But somewhere along the way, they turned into performance theater. We spend more time preparing to talk about our work than actually doing the work.

23 minutes to refocus after interruptions (University of California, Irvine study)

Every standup interrupts your flow state. And if you're context-switching between multiple projects, remembering what you did yesterday becomes genuine cognitive work.

The worst part? Most standup updates are already documented somewhere. In your commits. Your PRs. Your task tracker. Your Slack messages. You're just manually re-typing information that already exists.

What I Built: A Standup That Writes Itself

I wanted a system that:

Tracks my work automatically throughout the day
Generates accurate standup updates without manual input
Surfaces actual blockers, not generic status
Takes less than 30 seconds to review and send

Here's what I ended up with:

Step 1: Capture Everything Automatically

The foundation is automatic work capture. I needed a system that tracks:

Code commits and PR activity
Task updates and completions
Meeting notes and decisions
Research and documentation

I started using Pieces for Developers because it captures my workflow context automatically. When I'm coding, it tracks my commits and branches. When I'm in meetings, it captures notes and action items. When I'm researching, it saves relevant snippets and links.

The key is passive capture. I don't have to remember to log anything. It just happens in the background.

Step 2: Connect Your Work Sources

Most developers work across tons of tools:

GitHub/GitLab for code
Jira/Linear for tasks
Slack for communication
Notion/Confluence for docs

Your standup automation needs to pull from all of them. I set up integrations so my system knows:

Which PRs I opened or reviewed
Which tickets moved to "In Progress" or "Done"
Which meetings I attended and what was discussed
Which documentation I created or updated

This creates a complete activity log without manual tracking.

Step 3: Generate Smart Summaries

Raw activity logs are useless for standups. You need intelligent summarization that:

Groups related work together
Highlights completed items
Surfaces actual blockers
Formats for quick reading

I use Pieces Copilot to transform my activity log into standup format. It knows to:

Combine multiple commits into "Implemented feature X"
Flag PRs waiting on review as potential blockers
Identify patterns like "spent 3 hours debugging Y"
Format everything in past/present/future tense

The output looks like this:

**Yesterday:**
- Completed authentication refactor (PR #234)
- Fixed critical bug in payment processing
- Reviewed 3 PRs from team members

**Today:**
- Finishing API documentation
- Starting work on user dashboard redesign
- Team sync at 2pm

**Blockers:**
- Waiting on design approval for dashboard mockups
- Need database migration review from DevOps

Step 4: Review and Send in Seconds

Automation doesn't mean zero human involvement. I spend 30 seconds each morning reviewing the generated standup to:

Verify accuracy
Add context if needed
Adjust priorities
Flag urgent items

Then I copy-paste into Slack or our standup tool. Done.

The Results: Time Saved and Better Communication

After three months of automated standups, here's what changed:

65 minutes saved per week on standup prep and delivery

That's over 50 hours per year I'm not spending on status updates.

But the time savings aren't even the best part. The quality of my standups improved:

Before automation:

Generic updates like "worked on tickets"
Forgot to mention blockers until too late
Inconsistent detail level
Missed cross-team dependencies

After automation:

Specific, actionable updates with PR links
Blockers surfaced immediately
Consistent format and detail
Better visibility for stakeholders

My team lead actually commented that my standups became the most useful on the team. Not because I'm doing more work, but because the updates are more accurate and actionable.

Common Frustrations (And How to Fix Them)

"My work doesn't fit into yesterday/today/blockers format"

Fair point. Customize your template. Some teams use:

Focus areas
Progress metrics
Help needed
Wins and learnings

The automation adapts to whatever format you need. The key is having a consistent structure.

"I work on too many projects to track automatically"

This is actually more reason to automate. Context-switching makes manual standups even harder.

Use Long-Term Memory in Pieces to maintain separate contexts for each project. When you switch projects, your standup automation knows which context to pull from.

"My manager wants more detail than automated summaries provide"

Automation generates the baseline. You add the narrative.

I use the automated standup as my outline, then add 1-2 sentences of context about why something matters or how I approached a problem. Still takes less than 2 minutes total.

"What about remote/async teams?"

Async standups are perfect for automation. Instead of a synchronous meeting, everyone posts their update to a shared channel.

With automation, you can post your standup at any time without scrambling to remember what you did. Some teams even automate the posting itself on a schedule.

How to Start Automating Your Standups Today

You don't need to build a custom system from scratch. Here's how to start:

Option 1: Use Pieces for Developers (What I Use)

Install Pieces and connect your development tools
Let it capture your workflow for a few days
Use Copilot to generate standup summaries
Review and send

The Long-Term Memory feature maintains context across sessions, so your standups reflect your actual work patterns.

Option 2: Build Your Own Integration

If you want full control, build a custom integration:

# Pseudo-code for standup automation
def generate_standup():
    # Pull from Git
    commits = get_git_commits(since="yesterday")

    # Pull from task tracker
    tasks = get_completed_tasks(since="yesterday")

    # Pull from calendar
    meetings = get_meetings(date="yesterday")

    # Generate summary
    standup = format_standup({
        "yesterday": summarize_work(commits, tasks, meetings),
        "today": get_planned_work(),
        "blockers": identify_blockers()
    })

    return standup

Connect to GitHub API, Jira API, and Google Calendar API. Use an LLM API for summarization.

Option 3: Start Manual, Automate Incrementally

Not ready for full automation? Start with these habits:

Keep a daily work log (even just bullet points)
Use a template for consistency
Track blockers in real-time, not retrospectively
Review your log before standup instead of trying to remember

Then gradually add automation as you identify repetitive patterns.

The Bigger Picture: Automation Frees You to Focus

Automating standups isn't just about saving 15 minutes. It's about removing cognitive overhead.

When you're not mentally tracking "what did I do yesterday for standup," you can focus completely on the problem you're solving right now. No background anxiety about forgetting to mention something important.

"I used to dread standups because I could never remember everything I worked on. Now my standup writes itself and I actually look forward to sharing progress." — Me, three months into automation

The best part? This approach works for other recurring status updates too:

Weekly team reports
Sprint retrospectives
Monthly progress summaries
Performance review prep

Once you have automatic work capture, generating any status update becomes trivial.

Your Turn: What's Stopping You?

If you're still manually writing standups every morning, you're wasting time and mental energy on something that can be automated.

The tools exist. The integrations work. The time savings are real.

Start small:

Pick one work source to track automatically (start with Git commits)
Use a template for your standups
Add more automation incrementally

Or go all-in with a tool like Pieces that handles the whole workflow.

Either way, stop treating standup prep like it's valuable work. Automate it and focus on what actually matters: building great software.

What's your biggest standup frustration? Drop a comment—I'd love to hear how other devs are handling this.

Want to see how I automated my entire development workflow? Check out my other posts on developer productivity or try Pieces for Developers to start automating your standups today.

productivity automation devtools agile

Your Go-To Daily Standup Agenda and How to Actually Make It Work

Pieces 🌟 — Wed, 26 Nov 2025 16:00:32 +0000

If you've ever zoned out during a daily standup or wondered why you're even there, you're not alone. I've seen how a solid standup can transform a team's energy and output, but I've also watched these meetings turn into soul-crushing rituals where everyone just recites what they did yesterday while secretly checking Slack.

This guide is for anyone who wants standups that actually help - whether you're new to agile, leading a remote team, or just tired of wasting 15 minutes every morning on status theater.

What Is a Standup Meeting? (And Why Do We Still Do Them?)

Let's start with the basics: a standup is a short, focused daily meeting where team members quickly share what they're working on, what's next, and what's blocking them. The idea is to keep everyone aligned, surface blockers early, and move work forward without getting bogged down in status updates or side conversations.

But here's the truth: the value of a standup isn't in the ritual itself. It's in the clarity, accountability, and momentum it creates. When done right, a team standup is the heartbeat of a high-performing group. When done wrong, it's 15 minutes of your life you'll never get back.

Why Most Daily Standups Miss the Mark

Daily standups have been a staple of team workflows for years, but they're often more performative than productive. Instead of fostering real communication, these meetings devolve into rote status updates. Team members recite yesterday's tasks or scramble to sound busy while deeper blockers and real collaboration go unaddressed.

It becomes a kind of "productivity theater" where the focus shifts from building something meaningful to justifying your existence each morning. You know the feeling: you're halfway through your update and realize nobody's actually listening because they're mentally rehearsing their own lines.

The Common Frustrations

Here's what I've seen go wrong in tons of standups:

Status updates nobody needs: "Yesterday I worked on ticket 1234, today I'll work on ticket 1235" tells the team nothing useful
Hidden blockers: Real problems don't surface because people don't want to "waste everyone's time"
Meeting drift: What should take 10 minutes stretches to 25 because someone starts debugging in the standup
Disengagement: Half the team is on mute, clearly multitasking, because they know their turn won't come for another 8 minutes
Context switching pain: You're deep in flow, then boom - standup time, and it takes forever to get back into your work

At Pieces, we saw this firsthand and decided there had to be a better way. We built an AI memory tool that automatically captures your work context across code, docs, chats, and more, so you don't have to manually prepare or recall every detail for your standup.

By sharing routine updates asynchronously and letting the AI surface what matters, we improved standups from a daily performance into a focused, action-oriented discussion. We use our meetings to solve real problems, unblock each other, and move work forward.

Great Daily Standup Agenda: Classical vs. Modern Approach

When it comes to daily standups, most teams follow a familiar script. Here's how the classical approach compares to a more modern method, and why rethinking the format can be a game-changer for your team.

The Classical Standup Agenda

Traditionally, a daily standup agenda looks something like this:

Kickoff & Purpose (1 min)

Quick greeting, reminder of the meeting's goal

Round-Robin Updates (10 min)

Each person answers the classic questions:
- What did you do yesterday?
- What will you do today?
- Any blockers?

Blocker Triage (2-3 min)

Discuss and assign action items for any blockers

Announcements & Wrap-Up (1-2 min)

Share reminders, deadlines, or shoutouts

This format is simple and widely used, especially in agile and scrum teams. But over time, it often leads to repetitive status updates, disengagement, and meetings that drift off-topic.

A Modern Standup Approach

Here's how we've evolved our standup format to focus on what actually matters:

Step 1: Async Status Updates Before the Meeting

Everyone shares their routine updates (what they did, what's next, minor blockers) in a shared space before the meeting. This could be a Slack thread, a doc, or any tool that captures context automatically.

The key questions to answer asynchronously:

What progress did I make yesterday?
Where do I need the most help or support today?
Which accomplishments had the biggest impact on my teammates or the project?

This gives the team time to review, reflect, and prepare questions or feedback in advance. It also means no one has to scramble to remember yesterday's work on the spot, especially with tools that automatically capture your workflow.

Step 2: Focused, Synchronous Standup (5-10 min)

The live meeting is reserved for:

Surfacing critical blockers that need group input
Discussing urgent decisions or dependencies
Aligning on priorities for the day or sprint

No round-robin "what I did yesterday" unless it's relevant to the group. Tangents and deep dives are parked for after the meeting or handled asynchronously.

Step 3: Action-Oriented Wrap-Up

Confirm next steps, owners for blockers, and any follow-up conversations. You can use AI to help surface the most important blockers and challenges from your work context.

For example, asking "What was my biggest blocker or challenge, and how did I handle it?" can give you the full context without manual recall.

End on time - always. Respect your team's deep work time.

What Are the Benefits of This Approach?

Since the context is already available, you don't waste time catching up or rehashing yesterday's news. You can dive straight into the conversations that matter most, making every minute count.

Key Benefits

15 minutes saved per standup on average when using async updates

The biggest benefit is the respect it gives to deep work. Meetings become shorter and sharper, freeing up more of the day for actual progress. No one feels their focus is being hijacked by unnecessary chatter or drawn-out updates.

Specific improvements we've seen:

Reduced context switching: Fewer interruptions mean developers can maintain flow state longer
Better blocker visibility: Real problems surface earlier because there's time to think before the meeting
Increased engagement: When you're only meeting for what matters, people actually pay attention
More inclusive for remote teams: Async updates level the playing field for different time zones
Better documentation: Written updates create a searchable history of what happened and when

And the format keeps evolving. We regularly check in as a team, asking what's working and what could be better, and we're not afraid to tweak our format based on feedback or new challenges.

Practical Tips for Better Standups

Whether you're running a classic scrum standup or experimenting with async check-ins, here are some concrete ways to level up your meetings:

Keep It Focused

Time-box ruthlessly: Set a timer and stick to it
Park tangents: Create a "parking lot" for topics that need deeper discussion
Stay standing: If you're in person, actually stand - it naturally keeps things moving
One speaker at a time: No side conversations or debugging sessions

Make Status Updates Useful

Focus on outcomes, not activities: "Shipped the login feature" beats "Worked on authentication code"
Highlight dependencies: Call out when you need something from another team member
Be specific about blockers: "Waiting on API docs" is more actionable than "Blocked on backend"
Share context, not just task IDs: Help your team understand why your work matters

Use the Right Tools

Async-first communication: Use Slack, Discord, or dedicated standup tools for routine updates
Context capture: Tools that automatically track your work can save tons of prep time
Shared docs: Keep a running log of blockers and decisions for reference
Video for remote teams: Seeing faces builds connection, even in short meetings

Experiment and Iterate

Your standup format isn't set in stone. Try different approaches and see what works:

Rotate facilitators: Different perspectives can surface new improvements
Survey your team: Ask what's working and what's not every few weeks
Try different schedules: Maybe your team works better with standups at 2pm instead of 9am
Go async for a week: See if eliminating the meeting entirely improves or hurts communication

Make Your Standups Work for You

The key to effective standups is keeping them simple, focused, and action-oriented. Use the format that fits your team, stick to the essentials, and don't be afraid to tweak as you go.

Consider tools that help you capture context, automate summaries, and surface insights, so you can spend less time scrambling and more time building. The goal isn't to have perfect standups - it's to have standups that actually help your team ship better software faster.

What's your experience with daily standups? Have you found ways to make them more effective, or are you still stuck in status theater? Drop a comment below and let's discuss what's worked (or hasn't) for your team.

Ready to level up your standups? Try Pieces for free at your next meeting and see how automatic context capture can transform your team's clarity and momentum.

agile #productivity #teamwork #devtools

What's Ahead for Programmers: Tools Shaping the Future

Pieces 🌟 — Mon, 21 Oct 2024 18:14:37 +0000

In a recent talk, Chris Slee, a founder of Transform Labs in Columbus, Ohio, identified the tools developers will need to understand within the next 18 months. He posits that programming will move beyond the use of traditional copilots into using tools that reshape how developers interact with code. Slee described five key categories of tools with examples:

Frameworks
AI-friendly databases
Code writing tools
Pieces as a unique tool
Future frameworks

Each of these is discussed in one of the following sections.

1. Frameworks for Faster Code

As software development continues to evolve, programmers are increasingly turning to frameworks to accelerate their development processes. These frameworks, built upon existing code and libraries, provide a solid foundation for building, which saves time and effort. There are numerous commonly used frameworks such as TensorFlow and PyTorch.

DSPy is a Python library designed for digital signal processing (DSP). It offers a wide range of functions and tools for tasks such as filtering, Fourier transforms, and spectral analysis. By leveraging DSPy, developers can quickly implement complex DSP algorithms without having to write everything from scratch. It provides a user-friendly interface and efficient implementations, making it a popular choice for developers working on signal processing projects, even if the developers do not have extensive DSP experience.

Langchain is a framework focused on natural language processing (NLP). It provides a modular architecture and pre-built components for tasks like text generation, summarization, and question-answering. Langchain simplifies the process of building NLP applications, allowing developers to focus on the core functionality rather than low-level details.

There are several key benefits of using frameworks:

Faster development: Frameworks provide pre-built components and libraries, reducing the amount of code developers need to write. This enables developers to focus on core functionality and build applications more quickly.
Increased efficiency: By leveraging existing code, developers can avoid common pitfalls and improve the overall efficiency of their projects.
Improved code quality: Frameworks often adhere to best practices and coding standards, leading to higher-quality code.
Enhanced collaboration: Frameworks can facilitate collaboration among developers by providing a common foundation, shared vocabulary, and an active community that provides support and contributes to the framework's development. This can be invaluable for troubleshooting and learning new techniques.

Frameworks have become an essential tool for modern software development, enabling developers to build applications more efficiently and effectively. The best framework for a particular project depends on several factors, such as project requirements, language compatibility, dependencies on external libraries, ease of use, and community support.

By understanding the benefits and considerations of the specific requirements and tradeoffs of a project, developers can make informed decisions and choose the right tools to accelerate their projects. As the field of software development continues to evolve, even more innovative frameworks will emerge, further shaping the future of software development.

2. Efficient AI-Friendly Databases

The rapid advancements in artificial intelligence (AI) have created a surge in demand for databases that can effectively handle the unique storage requirements of AI applications. Traditional relational databases, while powerful, may struggle to efficiently store and process the massive datasets and complex computations often associated with AI.

To address these challenges, new database tools, particularly vector databases, have emerged as specialized solutions for AI workloads. Graphs and time series are also AI-friendly databases. Several leading AI-friendly databases have gained significant users in recent years:

Supabase: A low-code, open-source Firebase alternative that includes a powerful PostgreSQL database with built-in features for authentication, authorization, and storage. Supabase is well-suited for building real-time applications and APIs. It is designed to simplify the development of AI applications and can be easily integrated with popular AI frameworks.

Vertex AI: A fully managed platform from Google Cloud that provides a suite of tools for building, training, and deploying AI models. Vertex AI includes a high-performance vector database that is optimized for large-scale machine-learning workloads.

Pinecone: A cloud-based vector database service that is optimized for high-performance similarity search. It is designed to handle massive datasets, which means it is good for applications such as recommendation systems, search engines, and anomaly detection. It provides a flexible API and integrates well with popular AI frameworks.

These specialized databases offer several advantages for AI applications:

Efficient storage: They are optimized for storing and retrieving large datasets, which are common in AI workloads.
Fast search capabilities: Vector databases efficiently perform similarity searches, which are essential for many AI tasks.
Scalability: They can easily scale to handle increasing workloads as AI applications grow.
Integration with AI tools: Many AI-friendly databases integrate seamlessly with popular AI frameworks and tools, simplifying development and deployment.

The best choice of AI-friendly database depends on the specific requirements of the application, including:

Data type: Consider the type of data you will be storing (e.g., text, images, numerical data)
Search requirements: Determine the type of search queries you will perform (e.g., similarity search, exact match).
Performance requirements: Evaluate the performance requirements of your application, such as latency and throughput
Scalability needs: Consider how your data and workload may grow over time
Integration with existing systems: Consider how the database will integrate with your existing infrastructure and applications

AI-friendly databases are essential for building and scaling AI applications. By understanding the key features and benefits of these databases, you can make better decisions and be prepared for innovations in future databases designed specifically for AI systems.

3. Code-Writing Tools: Beyond Copilots

The landscape of code-writing tools is rapidly evolving, with new and innovative tools emerging to enhance developer productivity. While traditional code editors like VS Code remain popular, a new generation of code-writing tools is surpassing copilots by offering more advanced features and capabilities.

Aider: Aider is a comprehensive code-writing tool that can generate code from natural language prompts, explain code snippets, and suggest improvements. It supports a wide range of programming languages and integrates seamlessly with popular development environments.

Cline: Cline, previously known as Claude Dev, is a powerful AI-powered code completion tool that can suggest code snippets and entire functions based on the context of your code, and it can do AI code refactoring. It can help developers write cleaner, more maintainable code and improve their overall productivity.

01-Engineer: 01-Engineer is a versatile code-writing tool that can generate code from natural language prompts, explain code snippets, and provide debugging AI assistance. It is well-suited for data science and machine learning tasks.

Pythagora: Pythagora is a code generation tool that specializes in Python development. It can automatically generate Python code based on natural language descriptions, making it a valuable tool for data scientists or developers who are new to Python or need to quickly prototype new features.

Supermaven: Supermaven is a cloud-based development platform that offers a variety of tools and services for building, reviewing, testing, and deploying applications in several languages. It is designed to be highly customizable and can be tailored to the specific needs of individual developers.

Entire posts have been written about the benefits of using Pieces and other, less advanced copilot tools. This is a brief summary of the benefits of code-writing tools in general.

Increased productivity: Code writing tools can significantly accelerate the development process by automating repetitive tasks and suggesting code snippets.
Improved code quality: These tools can help developers write cleaner, more efficient, and more maintainable code.
Reduced errors: These tools can help identify and prevent common coding errors, reducing the risks of bugs and defects.
Reduced learning curve: For new developers, code writing tools can help reduce the learning curve by providing suggestions and explanations.
Enhanced collaboration: Some of these tools offer features that facilitate collaboration among development teams, such as code sharing and review.

As the field of artificial intelligence continues to advance, we can expect to see even more sophisticated code-writing tools emerge. These tools have the potential to revolutionize the way developers work, making them more productive, efficient, and creative.

4. Pieces for Developers

Slee called out Pieces as a "special thing" because its retrieval of similar code provides comparisons of other people's programming and "allows me to talk about code that somebody else wrote."

Pieces is also unique because of its innovative approach to context retention, knowledge repository, and collaboration functionality.

Context retention: Pieces will answer questions that no other software can, such as "What was the function I was working on last Wednesday?" The repository maintains a parallel workstream that maintains context information about the work done while it is active.
Knowledge repository: Pieces' plugins are available in various IDEs, browsers, and communication tools like Microsoft Teams. These all connect to the Pieces OS central repository, which means the developer can ask a question in the browser about work done earlier in the IDE.
Reuse and Collaboration: Pieces stores information that enriches stored code for reuse and sharing, such as who sent it to you or what website it came from. The code is organized and enhanced with information that makes it easy to understand the code, even when the code was written long ago or by someone else.

The context workflow and the enrichment information in Pieces is a game-changer. Overall, Pieces offers a unique and powerful solution for code reuse and collaboration. It provides developers with a comprehensive context workflow that can significantly improve their productivity and efficiency.

5. Emerging Frameworks and Tools:

The software development landscape is constantly evolving, with new frameworks and tools emerging to address the changing needs of developers. These are the futuristic tools that Slee included in his list.

Vercel V0: This generative UI framework, developed by Vercel, leverages machine learning to automatically generate code from design files (e.g., Figma, Sketch). Developers can describe their desired interface elements, and V0 automatically generates code snippets or prototypes.

Cursor: Cursor is a fork of VS Code that provides a natural language interface for a developer to write code by talking with an AI. It includes numerous tools for understanding, finding, refactoring, debugging, and writing code.

Zed AI: This open-source framework allows developers to create user interfaces with voice control capabilities. It provides pre-built components for tasks like computer vision, natural language processing, and machine learning. It offers tools for training machine learning models, integrating AI functionality into existing applications, and deploying AI systems.

FastBuilder AI: Another open-source framework, FastBuilder AI empowers developers to build web applications and APIs faster. It offers tools for automatic code generation and deployment, streamlining the development workflow.

Pear: This open-source AI platform focuses on simplifying web development for non-programmers. Pear enables users to create web pages using natural language commands with the intent to simplify and automate repetitive front-end development tasks. It offers tools for building reusable UI components, managing design systems, and streamlining the front-end development process.

The development landscape is constantly evolving, and these emerging tools showcase the innovative approaches being taken to improve developer productivity and functionality. The emergence of these innovative frameworks and tools suggests a future where development processes are faster, more collaborative, and more accessible. As these tools mature and gain wider adoption, we can expect to see a significant shift in the way software is designed and built.

Conclusion

Everything listed so far is predominantly software to be installed on a local machine. If two developers want to work on a function, they each would have to install it on their local machine and then sync their work through some type of repository.

In contrast, there is a new set of products coming online that are in the cloud. There is no need to install Visual Studio or other local software. Programming is moving away from using local machines to write code, and these innovations will strongly impact developers.

Consequently, it is crucially important to “stay ahead of the curve” and be aware of what is going to happen to programming in the near future. AI-based flexible tools such as Pieces will be able to go into the future, but static traditional tools will be left behind as time passes and the environment changes.

AI Transformation: How AI is Reshaping Enterprise Development

Pieces 🌟 — Mon, 21 Oct 2024 17:48:46 +0000

AI has become a major focus for business owners as we clearly see that the technology is gaining traction. Foundry’s 2024 CIO Tech Priorities study found that 89% of IT decision-makers are exploring, testing, or actively using AI technologies, a big jump from 72% in 2023.

64% of these decision-makers expect AI and machine learning to reshape business operations in the next three to five years, a significant rise from 39% who thought that just a year ago.

IT leaders see AI revolutionizing their organizations primarily through process automation, increased efficiency, and improved customer experiences. AI delivers benefits at a scale capable of driving transformation, not small improvements. Its impact has reached nearly every industry and is set to play a pivotal role in shaping the future of technology.

What Exactly is AI Transformation?

AI transformation (or sometimes referred to as AI digital transformation) is a strategic process where businesses use artificial intelligence in their operations, products, and services, driving innovation, efficiency, and growth. By leveraging a variety of AI models and technologies, organizations optimize workflows, make businesses more adaptive, and keep evolving.

Through machine learning, deep learning, and technologies like computer vision, natural language processing, and generative AI, companies can develop systems that:

Automate repetitive administrative tasks
Modernize applications and IT processes through code generation
Provide data-driven insights for better decision-making using advanced analytics
Continuously improve accuracy and performance by "learning" from data
Enhance customer experiences with personalization and AI chatbots

With rapid advancements in AI, embracing AI transformation is increasingly crucial for long-term business success. A report from the IBM Institute for Business Value, Augmented Work for an Automated, AI-Driven World, highlights that companies integrating AI into their transformation processes consistently outperform their competitors.

Unlike simply applying new technologies to replicate current processes, AI transformation is a more comprehensive initiative. A robust AI strategy for business transformation can introduce entirely new business methods, boost productivity, and promote sustainable growth. However, scaling AI often requires shifts in business strategies and company culture to fully embrace the technology's potential.

How is the AI Transformation Changing Businesses?

AI is important to the future of business and changing conventional practices to introduce new levels of efficiency and innovation. With AI and machine learning, businesses can analyze large data volumes, spot patterns, and deliver recommendations to improve decision-making.

This ability to process data faster than humans not only saves time, but also opens up new opportunities by enabling faster and more informed decisions.

One good example is how AI is going to change customer interaction. For example, ChatGPT, powered by generative AI, will allow companies to respond promptly and precisely to customer queries, thus reaching higher satisfaction levels.

AI-driven chatbots form one of the many ways in which this technology is used in reshaping operations, with applications even more general across a host of industries.

Let’s dive into how AI is helping businesses maintain their competitive edge:

Data-Driven Insights: Allow businesses to look through datasets faster and more accurately, using raw information to deliver actionable insights. Companies will be able to understand customer behavior better, optimize operations, and see market trends, which leads to more strategic decision-making.
Automation: Efficiency is bolstered, with the automation of tasks by the use of AI technologies like RPA and Machine Learning, to such an extent that manual input is considerably reduced. On this count, it will free staff for high-value, creative tasks, innovating more inside the organization.
Personalization: AI-driven algorithms analyze customer data and offer personal experiences, from product recommendations to marketing messages. This approach enhances satisfaction and loyalty and increases revenue by tailoring interactions to individual preferences.
Predictive Analytics: Predicting upcoming trends by examining past data, allowing for a proactive approach to market changes, dangers, and possibilities. This enables companies to make proactive decisions that minimize risks and optimize growth opportunities.
Enhanced Customer Support: Artificial intelligence virtual assistants and chatbots can manage customer service round the clock, offer fast responses, solve problems, and even finalize purchases. This improves customer experiences and lowers operational costs, reducing the need for human involvement.
Product Innovation: Helps in making innovative concepts by checking out market demands, identifying deficiencies, and producing fresh concepts for new products. This results in shorter product development cycles, faster time-to-market, and a more robust competitive edge.
Supply Chain Optimization: It makes supply chains easier to manage by forecasting demand, managing inventory efficiently, and improving logistics. Enhancements reduce costs and prevent disruptions, therefore improving general customer service and making supply chains more agile and responsive.

The Path to Digital Transformation and AI in Enterprises

Turning AI from a novelty into a powerful tool that drives enterprise success requires a structured approach. The process starts by identifying clear business objectives and assembling a dedicated team of data experts and business leaders.

These individuals must possess strong technical skills and be collaborative, detail-oriented, and capable of creating a comprehensive data ecosystem.

Architects are essential for designing, connecting, and securing this system, while a dedicated team ensures smooth operations and monitoring. To turn AI from a buzzword into a valuable business asset, the following steps are critical:

Understand Business Objectives

The journey begins with a thorough evaluation of the company’s goals, capabilities, scalability needs, and compliance requirements. AI must be aligned with these factors to deliver meaningful value from the outset.

Cultural and Ethical Integration

Smooth adoption of AI is not just a technical challenge, there are also cultural and ethical issues to consider. Senior executives should be involved in brainstorming on where AI can assist. Also, ethical issues need to be considered, including how human judgment will oversee AI and how it would supplement a team’s work without displacing jobs.

Identifying Data Sources

A crucial step is locating all potential data sources within the organization, whether they are linked or operate independently. This includes data from systems like CRMs, ERPs, customer service logs, emails, images, and other key records.

Building a Comprehensive Data Repository

Creating a structured data set that aligns with business goals is vital for training AI models. This requires careful planning around building a data lake, data warehouse, or data lakehouse, along with data migration, modernization, storage considerations, and ensuring that the data meets security and compliance standards.

Training AI with Proprietary Data

While using pre-trained models such as ChatGPT can be beneficial, the true value of AI emerges when it is trained on an organization’s proprietary data. This allows for the automation and augmentation of repetitive tasks, freeing employees to focus on more complex and value-driven activities.

Ethical and Environmental Challenges of AI Transformation

While AI offers tremendous benefits, its adoption brings significant ethical challenges. One pressing concern is algorithmic bias. For example, facial recognition technologies, like those used by law enforcement, have faced criticism for racial biases that have led to false arrests. The National Institute of Standards and Technology (NIST) found that these algorithms were more likely to misidentify people of color, raising concerns about fairness and discrimination.

Moreover, AI’s environmental impact cannot be ignored. Training large AI models requires enormous computational power, which consumes vast amounts of energy. According to a study by the University of Massachusetts, training a single deep-learning model can emit as much carbon dioxide as five cars over their entire lifetimes.

Considering the environmental footprints, companies today are finding ways to offset the effects of AI by powering such data centers with renewable energy. For example, Google plans to run all its data centers on 100% renewable energy by 2030.

Businesses must meet these challenges by showing responsibility when developing and deploying AI. Best practices for ethical AI include routine audits to prevent biases, focusing on privacy protection, and embracing sustainable computing. Microsoft is a good example and has started an AI for Earth program to use AI to solve environmental problems like climate change, biodiversity, and water sustainability.

Prepare Your Business for AI Transformation With Pieces

AI transformation continues to reshape industries and drive growth. Projections estimate AI could contribute $15.7 trillion to the global economy by 2030. As more AI technologies integrate into manufacturing, healthcare, and other industries, the impact on the economy will be big. Embracing AI early positions businesses to lead, driving new standards for excellence.

We are prepared to assist you in integrating AI technology with customized solutions to boost productivity, enhance customer engagement, and promote ethical and sustainable AI utilization.