Skip to main content

STM (Short-Term Memory) Functions

Store and retrieve data in S3 with automatic organization by organization and environment. STM provides hierarchical storage for workflow state, session data, and temporary files.

Overview​

The STM integration provides three functions:

  • stmStore() - Store a value with metadata path
  • stmRetrieve() - Retrieve a specific file
  • stmRetrieveAll() - Retrieve all files under a metadata prefix

How STM Works​

Storage Structure​

Data is stored in S3 with an automatically generated path:

s3://{bucket}/{organizationId}/{environmentId}/{metadata[0]}/{metadata[1]}/.../filename

Example:

await stmStore(['sessions', 'user-123', 'chat'], { history: [...] }, 'state.json');

Creates: s3://rw-stream-stm/{orgId}/{envId}/sessions/user-123/chat/state.json

Automatic Context​

The organizationId and environmentId are automatically extracted from the incoming message object. You don't need to specify them.

Functions​

stmStore​

Store a value to S3 short-term memory.

Signature:

async function stmStore(
  metadata: string[],
  value: any,
  filename: string
): Promise<{
  success: boolean;
  bucket: string;
  key: string;
}>

Parameters:

ParameterTypeDescription
metadatastring[]Array of folder names for hierarchy
valueanyValue to store (objects are JSON stringified)
filenamestringName of the file

Returns: Object with storage details

Example:

const result = await stmStore(
  ['users', 'user-123'],
  { name: 'John', score: 100 },
  'profile.json'
);

print('Stored at:', result.key);
// Output: Stored at: {orgId}/{envId}/users/user-123/profile.json

stmRetrieve​

Retrieve a value from S3 short-term memory.

Signature:

async function stmRetrieve(
  metadata: string[],
  filename: string
): Promise<any | null>

Parameters:

ParameterTypeDescription
metadatastring[]Array of folder names for hierarchy
filenamestringName of the file to retrieve

Returns: The stored value (parsed from JSON if applicable) or null if not found

Example:

const profile = await stmRetrieve(['users', 'user-123'], 'profile.json');

if (profile) {
  print('Name:', profile.name);
  print('Score:', profile.score);
} else {
  print('Profile not found');
}

stmRetrieveAll​

Retrieve all values under a metadata prefix.

Signature:

async function stmRetrieveAll(
  metadata: string[]
): Promise<Array<{
  key: string;
  filename: string;
  value: any;
}>>

Parameters:

ParameterTypeDescription
metadatastring[]Array of folder names for prefix

Returns: Array of objects with key, filename, and value

Example:

// Get all files in the user-123 folder
const allFiles = await stmRetrieveAll(['users', 'user-123']);

for (const file of allFiles) {
  print('File:', file.filename);
  print('Value:', JSON.stringify(file.value));
}

Complete Examples​

Example 1: Chat Session State​

Store and retrieve chat conversation history:

Store Chat Message (Action Script):

// Get existing chat history
const sessionId = message.sessionId || message.correlationId;
let chatHistory = await stmRetrieve(['chat', sessionId], 'history.json');

// Initialize if first message
if (!chatHistory) {
  chatHistory = {
    sessionId: sessionId,
    startedAt: new Date().toISOString(),
    messages: []
  };
}

// Add new message
chatHistory.messages.push({
  role: 'user',
  content: message.content,
  timestamp: new Date().toISOString()
});

// Add AI response if available
const aiResponse = await latestPromptResponse();
if (aiResponse) {
  chatHistory.messages.push({
    role: 'assistant',
    content: aiResponse,
    timestamp: new Date().toISOString()
  });
}

// Store updated history
await stmStore(['chat', sessionId], chatHistory, 'history.json');
print('Chat history updated. Messages:', chatHistory.messages.length);

Example 2: User Preferences​

Track user settings across workflow executions:

Action Script:

const userId = message.userId;

// Load existing preferences or create defaults
let prefs = await stmRetrieve(['users', userId], 'preferences.json');

if (!prefs) {
  prefs = {
    userId: userId,
    createdAt: new Date().toISOString(),
    timezone: 'UTC',
    notifications: true,
    theme: 'dark'
  };
}

// Update based on message
if (message.updatePreferences) {
  prefs = { ...prefs, ...message.updatePreferences };
  prefs.updatedAt = new Date().toISOString();
}

// Save preferences
await stmStore(['users', userId], prefs, 'preferences.json');

// Make preferences available to downstream nodes
output = { preferences: prefs };

Example 3: Rate Limiting​

Track API calls to implement rate limiting:

Action Script:

const userId = message.userId;
const windowMinutes = 60;
const maxRequests = 100;

// Get current rate limit data
let rateData = await stmRetrieve(['ratelimit', userId], 'current.json');
const now = Date.now();

// Initialize or reset if window expired
if (!rateData || now - rateData.windowStart > windowMinutes * 60 * 1000) {
  rateData = {
    windowStart: now,
    count: 0
  };
}

// Check limit
if (rateData.count >= maxRequests) {
  print('Rate limit exceeded for user:', userId);
  output = {
    allowed: false,
    remainingRequests: 0,
    resetAt: new Date(rateData.windowStart + windowMinutes * 60 * 1000).toISOString()
  };
} else {
  // Increment and save
  rateData.count++;
  await stmStore(['ratelimit', userId], rateData, 'current.json');
  
  output = {
    allowed: true,
    remainingRequests: maxRequests - rateData.count,
    currentCount: rateData.count
  };
}

Example 4: Aggregating Sports Data​

Collect game events over time:

Action Script:

const gameId = message.gameId;
const eventType = message.eventType;

// Get current game data
let gameData = await stmRetrieve(['games', gameId], 'events.json');

if (!gameData) {
  gameData = {
    gameId: gameId,
    startedAt: new Date().toISOString(),
    events: [],
    stats: {
      touchdowns: 0,
      fieldGoals: 0,
      interceptions: 0
    }
  };
}

// Add event
gameData.events.push({
  type: eventType,
  data: message,
  timestamp: new Date().toISOString()
});

// Update stats
if (eventType === 'touchdown') gameData.stats.touchdowns++;
if (eventType === 'field_goal') gameData.stats.fieldGoals++;
if (eventType === 'interception') gameData.stats.interceptions++;

// Save updated game data
await stmStore(['games', gameId], gameData, 'events.json');

print('Game', gameId, '- Events:', gameData.events.length);
output = gameData.stats;

Example 5: Multi-File Storage​

Store related data in separate files:

Action Script:

const reportId = message.reportId;
const reportData = message.data;

// Store report in multiple files for organization
await stmStore(['reports', reportId], {
  id: reportId,
  title: reportData.title,
  createdAt: new Date().toISOString()
}, 'metadata.json');

await stmStore(['reports', reportId], {
  summary: reportData.summary,
  details: reportData.details
}, 'content.json');

await stmStore(['reports', reportId], {
  charts: reportData.charts,
  tables: reportData.tables
}, 'visuals.json');

print('Report stored in 3 files');

// Later, retrieve all parts
const allParts = await stmRetrieveAll(['reports', reportId]);
print('Retrieved parts:', allParts.length);

for (const part of allParts) {
  print('  -', part.filename);
}

Example 6: Workflow Checkpointing​

Save workflow state for resumption:

Action Script:

const workflowRunId = message.workflowRunId || message.correlationId;
const currentStep = message.currentStep || 'initial';

// Load checkpoint
let checkpoint = await stmRetrieve(['workflows', workflowRunId], 'checkpoint.json');

if (!checkpoint) {
  checkpoint = {
    runId: workflowRunId,
    startedAt: new Date().toISOString(),
    steps: [],
    currentStep: currentStep,
    state: {}
  };
}

// Update checkpoint
checkpoint.steps.push({
  step: currentStep,
  completedAt: new Date().toISOString(),
  success: true
});
checkpoint.currentStep = currentStep;
checkpoint.lastUpdated = new Date().toISOString();

// Store workflow-specific state
if (message.stateUpdate) {
  checkpoint.state = { ...checkpoint.state, ...message.stateUpdate };
}

// Save checkpoint
await stmStore(['workflows', workflowRunId], checkpoint, 'checkpoint.json');

print('Checkpoint saved. Steps completed:', checkpoint.steps.length);
output = { checkpoint: checkpoint };

Troubleshooting​

"message.organizationId and message.environmentId are required"​

Cause: The incoming message doesn't have org/env IDs

Solution:

  • Ensure the message is properly routed through the platform
  • Check that the receiver is including these fields

"metadata must be an array of strings"​

Cause: Invalid metadata parameter

Solution:

// Wrong
await stmStore('users/123', data, 'file.json');

// Correct
await stmStore(['users', '123'], data, 'file.json');

"filename is required and must be a string"​

Cause: Missing or invalid filename

Solution:

// Include .json extension for JSON data
await stmStore(['data'], value, 'data.json');

Returns null when data should exist​

Causes:

  1. Wrong metadata path
  2. Wrong filename
  3. Different organization or environment

Debug:

print('Looking for:', JSON.stringify({
  org: message.organizationId,
  env: message.environmentId,
  path: ['users', userId],
  file: 'data.json'
}));

const result = await stmRetrieve(['users', userId], 'data.json');
print('Result:', result);

Best Practices​

  1. Use meaningful metadata paths - Organize data logically (e.g., ['users', id], ['sessions', sessionId])

  2. Include IDs in paths - Use unique identifiers to separate data:

    // Good
    await stmStore(['chat', sessionId], data, 'history.json');

    // Bad - will overwrite other users' data
    await stmStore(['chat'], data, 'history.json');

  3. Handle missing data - Always check for null results:

    const data = await stmRetrieve(['key'], 'file.json') || defaultValue;

  4. Use JSON filenames - Add .json extension for serialized data

  5. Keep values reasonable size - Large objects take longer to store/retrieve

  6. Clean up old data - Implement TTL logic if needed (data doesn't auto-expire)